From 5ddf381eed710473eb24bd51edceb3e5bf7441cc Mon Sep 17 00:00:00 2001 From: Vito Cappello Date: Sat, 22 Aug 2026 16:44:47 -0400 Subject: [PATCH] fix(channels): preserve gateway context for inbound turns (#127962) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(channels): preserve gateway context for inbound turns * fix(plugin-sdk): type bound channel reply dispatcher * fix(channels): carry bound reply dispatchers * fix(channels): keep reply carrier internal * fix(auto-reply): consolidate dispatcher type imports * fix(channels): keep reply dispatch typing internal * fix(channels): derive inbound reply dispatcher types from turn plan Keep ChannelRuntimeSurface identical to main: adapters read the bound dispatchReplyFromConfig through the existing PluginRuntime["channel"] wiring type and derive its type from the public ChannelInboundTurnPlan contract, so the compatibility surface no longer grows a Gateway-bound reply member. Co-authored-by: Cursor * fix(channels): hoist bound channel runtime casts for assertion ratchet The assertion-safety ratchet collects SAFETY comments with a plain ts.Scanner pass, which desyncs at the first template literal with a substitution — comments after that point in a file are invisible, so deep casts in monitor files cannot be SAFETY-covered. Hoist one shared cast per call-site scope (absorbing the pre-existing buildContext casts) and retype Discord's internal channelRuntime chain as PluginRuntime["channel"] so dispatch reads need no assertion. Co-authored-by: Cursor * test(channels): verify gateway dispatcher ownership Co-authored-by: VACInc <3279061+VACInc@users.noreply.github.com> * test(codex): assert sequenced node process notifications Co-authored-by: VACInc <3279061+VACInc@users.noreply.github.com> * docs(changelog): preserve release-owned root changelog Channel Gateway tools preserve the owning dispatcher for Telegram, Discord, iMessage, Signal, and WhatsApp so terminal tools remain available. Thanks @VACInc. Co-authored-by: VACInc <3279061+VACInc@users.noreply.github.com> * test(codex): avoid shadowing process notification bindings Co-authored-by: VACInc <3279061+VACInc@users.noreply.github.com> --------- Co-authored-by: VACInc <3279061+VACInc@users.noreply.github.com> Co-authored-by: Cursor Co-authored-by: Peter Steinberger --- config/assertion-safety-baseline.txt | 2 +- extensions/codex/src/node-exec-server.test.ts | 51 ++++++++++--------- .../src/monitor/agent-components.dispatch.ts | 2 + .../src/monitor/agent-components.types.ts | 2 + .../src/monitor/provider.interactions.ts | 5 +- extensions/discord/src/monitor/provider.ts | 7 +-- .../imessage/src/monitor/monitor-provider.ts | 7 ++- .../signal/src/monitor/event-handler.ts | 2 + .../telegram/src/bot-message-dispatch-turn.ts | 1 + ...bot-message-dispatch.pipeline-init.test.ts | 17 +++++++ .../src/bot-message-dispatch.test-harness.ts | 4 +- .../src/bot-message-dispatch.types.ts | 5 +- extensions/telegram/src/bot-message.ts | 7 ++- extensions/telegram/src/bot.types.ts | 5 ++ extensions/telegram/src/monitor.ts | 13 +++-- extensions/telegram/src/polling-session.ts | 2 + extensions/telegram/src/webhook.ts | 2 + extensions/whatsapp/src/auto-reply/monitor.ts | 9 +++- .../src/auto-reply/monitor/on-message.ts | 3 ++ .../monitor/process-message.test.ts | 11 +++- .../src/auto-reply/monitor/process-message.ts | 3 ++ .../reply/provider-dispatcher.test.ts | 6 ++- src/auto-reply/reply/provider-dispatcher.ts | 1 + .../reply/provider-dispatcher.types.ts | 6 ++- .../context-builder-callers.test.ts | 14 ++--- src/gateway/server-plugins.test.ts | 18 +++++++ 26 files changed, 149 insertions(+), 56 deletions(-) diff --git a/config/assertion-safety-baseline.txt b/config/assertion-safety-baseline.txt index 94b29e69c37b..65399f950c63 100644 --- a/config/assertion-safety-baseline.txt +++ b/config/assertion-safety-baseline.txt @@ -1297,7 +1297,7 @@ extensions/telegram/src/message-cache.ts 10 extensions/telegram/src/miniapp/command.ts 1 extensions/telegram/src/miniapp/init-data.ts 1 extensions/telegram/src/miniapp/routes.ts 1 -extensions/telegram/src/monitor.ts 4 +extensions/telegram/src/monitor.ts 3 extensions/telegram/src/network-config.ts 1 extensions/telegram/src/network-errors.ts 8 extensions/telegram/src/outbound-adapter.ts 5 diff --git a/extensions/codex/src/node-exec-server.test.ts b/extensions/codex/src/node-exec-server.test.ts index 6181227c3034..711f120067b3 100644 --- a/extensions/codex/src/node-exec-server.test.ts +++ b/extensions/codex/src/node-exec-server.test.ts @@ -115,6 +115,23 @@ async function readNodeResponse( return response.result as JsonRpcRecord; } +async function readNodeProcessNotifications( + frames: ReturnType, + processId: string, + count: number, +): Promise { + const matching = () => + frames.outbound.filter( + (message) => + String(message.method).startsWith("process/") && + (message.params as { processId?: string }).processId === processId, + ); + await vi.waitFor(() => expect(matching()).toHaveLength(count)); + return matching().toSorted( + (left, right) => (left.params as { seq: number }).seq - (right.params as { seq: number }).seq, + ); +} + afterEach(() => { vi.unstubAllEnvs(); }); @@ -364,14 +381,7 @@ describe("Codex paired-node exec-server", () => { }, }); await readNodeResponse(frames, 9); - await vi.waitFor(() => - expect(frames.outbound.some((message) => message.method === "process/closed")).toBe( - true, - ), - ); - const notifications = frames.outbound.filter((message) => - String(message.method).startsWith("process/"), - ); + const notifications = await readNodeProcessNotifications(frames, "node-proof", 3); expect(notifications.map((message) => message.method)).toEqual([ "process/output", "process/exited", @@ -492,24 +502,15 @@ describe("Codex paired-node exec-server", () => { }, }); expect(await readNodeResponse(frames, control.id + 1)).toEqual(control.result); - await vi.waitFor(() => - expect( - frames.outbound.some( - (message) => - message.method === "process/closed" && - (message.params as { processId?: string }).processId === control.processId, - ), - ).toBe(true), + const controlNotifications = await readNodeProcessNotifications( + frames, + control.processId, + 2, ); - expect( - frames.outbound - .filter( - (message) => - String(message.method).startsWith("process/") && - (message.params as { processId?: string }).processId === control.processId, - ) - .map((message) => message.method), - ).toEqual(["process/exited", "process/closed"]); + expect(controlNotifications.map((message) => message.method)).toEqual([ + "process/exited", + "process/closed", + ]); } const httpServer = createServer((_request, response) => { diff --git a/extensions/discord/src/monitor/agent-components.dispatch.ts b/extensions/discord/src/monitor/agent-components.dispatch.ts index 96124be017f1..c88fbb932d58 100644 --- a/extensions/discord/src/monitor/agent-components.dispatch.ts +++ b/extensions/discord/src/monitor/agent-components.dispatch.ts @@ -282,6 +282,8 @@ export async function dispatchDiscordComponentEvent(params: { accountId, route: { agentId, sessionKey }, ctxPayload, + // Forward the owning runtime's bound dispatcher into the turn plan; never invoked here. + dispatchReplyFromConfig: ctx.channelRuntime?.reply?.dispatchReplyFromConfig, record: { updateLastRoute: interactionCtx.isDirectMessage ? { diff --git a/extensions/discord/src/monitor/agent-components.types.ts b/extensions/discord/src/monitor/agent-components.types.ts index e224fdaa3282..e286c0797fa0 100644 --- a/extensions/discord/src/monitor/agent-components.types.ts +++ b/extensions/discord/src/monitor/agent-components.types.ts @@ -1,4 +1,5 @@ // Discord type declarations define plugin contracts. +import type { PluginRuntime } from "openclaw/plugin-sdk/channel-core"; import type { DiscordAccountConfig, OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; import type { ButtonInteraction, @@ -40,6 +41,7 @@ export type AgentComponentContext = { accountId: string; discordConfig?: DiscordAccountConfig; runtime?: import("openclaw/plugin-sdk/runtime-env").RuntimeEnv; + channelRuntime?: PluginRuntime["channel"]; token?: string; guildEntries?: Record; allowFrom?: string[]; diff --git a/extensions/discord/src/monitor/provider.interactions.ts b/extensions/discord/src/monitor/provider.interactions.ts index 690b7c774081..bd9b9b52e2de 100644 --- a/extensions/discord/src/monitor/provider.interactions.ts +++ b/extensions/discord/src/monitor/provider.interactions.ts @@ -1,6 +1,6 @@ // Discord provider module implements model/runtime integration. import { CHANNEL_APPROVAL_NATIVE_RUNTIME_CONTEXT_CAPABILITY } from "openclaw/plugin-sdk/approval-handler-adapter-runtime"; -import type { ChannelRuntimeSurface } from "openclaw/plugin-sdk/channel-contract"; +import type { PluginRuntime } from "openclaw/plugin-sdk/channel-core"; import { registerChannelRuntimeContext } from "openclaw/plugin-sdk/channel-runtime-context"; import type { DiscordAccountConfig, OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; import type { RuntimeEnv } from "openclaw/plugin-sdk/runtime-env"; @@ -55,7 +55,7 @@ export function createDiscordProviderInteractionSurface(params: { allowFrom: DiscordAccountConfig["allowFrom"]; dmPolicy: NonNullable; runtime: RuntimeEnv; - channelRuntime?: ChannelRuntimeSurface; + channelRuntime?: PluginRuntime["channel"]; abortSignal?: AbortSignal; createNativeCommand?: typeof createDiscordNativeCommand; }): { @@ -163,6 +163,7 @@ export function createDiscordProviderInteractionSurface(params: { allowFrom: params.allowFrom, dmPolicy: params.dmPolicy, runtime: params.runtime, + channelRuntime: params.channelRuntime, token: params.token, }, params.applicationId, diff --git a/extensions/discord/src/monitor/provider.ts b/extensions/discord/src/monitor/provider.ts index b9fee5aff4fb..ccd2cb3afeb4 100644 --- a/extensions/discord/src/monitor/provider.ts +++ b/extensions/discord/src/monitor/provider.ts @@ -309,6 +309,8 @@ export async function monitorDiscordProvider(opts: MonitorDiscordOpts = {}) { let earlyGatewayEmitter = gatewaySupervisor?.emitter; let onEarlyGatewayDebug: ((msg: unknown) => void) | undefined; try { + // SAFETY: Gateway startup supplies the full plugin channel runtime; the surface type is the minimal external view. + const pluginChannelRuntime = opts.channelRuntime as PluginRuntime["channel"] | undefined; const { commands, components, modals } = createDiscordProviderInteractionSurface({ cfg, discordConfig: discordCfg, @@ -328,7 +330,7 @@ export async function monitorDiscordProvider(opts: MonitorDiscordOpts = {}) { allowFrom, dmPolicy, runtime, - channelRuntime: opts.channelRuntime, + channelRuntime: pluginChannelRuntime, abortSignal: opts.abortSignal, createNativeCommand: discordProviderRuntime.createDiscordNativeCommand, }); @@ -442,8 +444,7 @@ export async function monitorDiscordProvider(opts: MonitorDiscordOpts = {}) { accountId: account.accountId, token, runtime, - buildContext: (opts.channelRuntime as PluginRuntime["channel"] | undefined)?.inbound - .buildContext, + buildContext: pluginChannelRuntime?.inbound.buildContext, setStatus: opts.setStatus, abortSignal: opts.abortSignal, botUserId, diff --git a/extensions/imessage/src/monitor/monitor-provider.ts b/extensions/imessage/src/monitor/monitor-provider.ts index 8aeae48628ca..45c461a1e5e3 100644 --- a/extensions/imessage/src/monitor/monitor-provider.ts +++ b/extensions/imessage/src/monitor/monitor-provider.ts @@ -1059,6 +1059,8 @@ export async function monitorIMessageProvider(opts: MonitorIMessageOpts = {}): P logVerbose, }) : undefined; + // SAFETY: Gateway startup supplies the full plugin channel runtime; the surface type is the minimal external view. + const pluginChannelRuntime = opts.channelRuntime as PluginRuntime["channel"] | undefined; const { ctxPayload, chatTarget, imessageTo } = await buildIMessageInboundContext({ cfg, accountService: imessageCfg.service, @@ -1069,8 +1071,7 @@ export async function monitorIMessageProvider(opts: MonitorIMessageOpts = {}): P historyLimit, groupHistories, dmHistory, - buildContext: (opts.channelRuntime as PluginRuntime["channel"] | undefined)?.inbound - .buildContext, + buildContext: pluginChannelRuntime?.inbound.buildContext, media: { facts: mediaAttachments, }, @@ -1269,6 +1270,8 @@ export async function monitorIMessageProvider(opts: MonitorIMessageOpts = {}): P sessionKey: decision.route.sessionKey, }, ctxPayload, + // Forward the owning runtime's bound dispatcher into the turn plan; never invoked here. + dispatchReplyFromConfig: pluginChannelRuntime?.reply?.dispatchReplyFromConfig, record: { updateLastRoute: !decision.isGroup && updateTarget diff --git a/extensions/signal/src/monitor/event-handler.ts b/extensions/signal/src/monitor/event-handler.ts index af15c1d4a8b5..2abb6bf0c73a 100644 --- a/extensions/signal/src/monitor/event-handler.ts +++ b/extensions/signal/src/monitor/event-handler.ts @@ -523,6 +523,8 @@ export function createSignalEventHandler(deps: SignalEventHandlerDeps) { accountId: route.accountId, route: { agentId: route.agentId, sessionKey: route.sessionKey }, ctxPayload, + // Forward the owning runtime's bound dispatcher into the turn plan; never invoked here. + dispatchReplyFromConfig: deps.channelRuntime?.reply?.dispatchReplyFromConfig, record: { updateLastRoute: !entry.isGroup ? { diff --git a/extensions/telegram/src/bot-message-dispatch-turn.ts b/extensions/telegram/src/bot-message-dispatch-turn.ts index 199283d54264..2cf125234f84 100644 --- a/extensions/telegram/src/bot-message-dispatch-turn.ts +++ b/extensions/telegram/src/bot-message-dispatch-turn.ts @@ -123,6 +123,7 @@ export async function runTelegramDispatchTurn(turn: Turn) { }, ctxPayload: context.ctxPayload, record: context.turn.record, + dispatchReplyFromConfig: turn.opts.dispatchReplyFromConfig, delivery: { deliverWithProviderMessageSending: async (payload, info) => await deliverReply(turn, payload, info), diff --git a/extensions/telegram/src/bot-message-dispatch.pipeline-init.test.ts b/extensions/telegram/src/bot-message-dispatch.pipeline-init.test.ts index c59925bbe34a..3e37b8600b0c 100644 --- a/extensions/telegram/src/bot-message-dispatch.pipeline-init.test.ts +++ b/extensions/telegram/src/bot-message-dispatch.pipeline-init.test.ts @@ -4,6 +4,7 @@ import { createContext, createRuntime, createStatusReactionController, + dispatchReplyWithBufferedBlockDispatcher, describeTelegramDispatch, dispatchWithContext, } from "./bot-message-dispatch.test-harness.js"; @@ -11,6 +12,22 @@ import type { TelegramMessageContext } from "./bot-message-dispatch.test-harness import { telegramInboundEventDelivery } from "./inbound-event-delivery.js"; describeTelegramDispatch("dispatchTelegramMessage pipeline-init", () => { + it("keeps the owning Gateway reply dispatcher on the assembled inbound turn", async () => { + const dispatchReplyFromConfig = vi.fn(); + + await dispatchWithContext({ + context: createContext(), + opts: { + token: "token", + dispatchReplyFromConfig, + } as Parameters[0]["opts"], + }); + + expect(dispatchReplyWithBufferedBlockDispatcher).toHaveBeenCalledWith( + expect.objectContaining({ dispatchReplyFromConfig }), + ); + }); + it("does not enter the reply pipeline after the durable owner aborts", async () => { const abortController = new AbortController(); abortController.abort(new Error("handler-timeout")); diff --git a/extensions/telegram/src/bot-message-dispatch.test-harness.ts b/extensions/telegram/src/bot-message-dispatch.test-harness.ts index f3293044bfa9..6dadc81db7e1 100644 --- a/extensions/telegram/src/bot-message-dispatch.test-harness.ts +++ b/extensions/telegram/src/bot-message-dispatch.test-harness.ts @@ -241,6 +241,7 @@ vi.mock("openclaw/plugin-sdk/channel-inbound", async (importOriginal) => { toolsAllow: resolved.toolsAllow, replyOptions: resolved.replyOptions, replyResolver: resolved.replyResolver, + dispatchReplyFromConfig: resolved.dispatchReplyFromConfig, }); return withTelegramTestSettledReceipt(dispatchResult); }, @@ -673,6 +674,7 @@ export async function dispatchWithContext(params: { textLimit?: number; turnAdoptionLifecycle?: Parameters[0]["turnAdoptionLifecycle"]; runtime?: Parameters[0]["runtime"]; + opts?: Parameters[0]["opts"]; }) { const bot = params.bot ?? createBot(); return await dispatchTelegramMessage({ @@ -685,7 +687,7 @@ export async function dispatchWithContext(params: { textLimit: params.textLimit ?? 4096, telegramCfg: params.telegramCfg ?? {}, telegramDeps: params.telegramDeps ?? telegramDepsForTest, - opts: { token: "token" }, + opts: params.opts ?? { token: "token" }, retryDispatchErrors: params.retryDispatchErrors, suppressFailureFallback: params.suppressFailureFallback, turnAdoptionLifecycle: params.turnAdoptionLifecycle, diff --git a/extensions/telegram/src/bot-message-dispatch.types.ts b/extensions/telegram/src/bot-message-dispatch.types.ts index b035127f8dad..f98d3a800cdb 100644 --- a/extensions/telegram/src/bot-message-dispatch.types.ts +++ b/extensions/telegram/src/bot-message-dispatch.types.ts @@ -35,7 +35,10 @@ export type DispatchTelegramMessageParams = { textLimit: number; telegramCfg: TelegramAccountConfig; telegramDeps?: TelegramBotDeps; - opts: Pick; + opts: Pick< + TelegramBotOptions, + "token" | "mediaMaxMb" | "ownerAgentId" | "dispatchReplyFromConfig" + >; retryDispatchErrors?: boolean; suppressFailureFallback?: boolean; /** diff --git a/extensions/telegram/src/bot-message.ts b/extensions/telegram/src/bot-message.ts index 62f4b7e321e7..6d47fdce09a2 100644 --- a/extensions/telegram/src/bot-message.ts +++ b/extensions/telegram/src/bot-message.ts @@ -72,7 +72,12 @@ type TelegramMessageProcessorDeps = Omit< buildContext?: typeof import("openclaw/plugin-sdk/channel-inbound").buildChannelInboundEventContext; opts: Pick< TelegramBotOptions, - "token" | "ownerAgentId" | "allowFrom" | "groupAllowFrom" | "replyToMode" + | "token" + | "ownerAgentId" + | "allowFrom" + | "groupAllowFrom" + | "replyToMode" + | "dispatchReplyFromConfig" >; }; diff --git a/extensions/telegram/src/bot.types.ts b/extensions/telegram/src/bot.types.ts index bdedd13d7292..a40590085ccf 100644 --- a/extensions/telegram/src/bot.types.ts +++ b/extensions/telegram/src/bot.types.ts @@ -1,10 +1,13 @@ // Telegram type declarations define plugin contracts. +import type { ChannelInboundTurnPlan } from "openclaw/plugin-sdk/channel-inbound"; import type { OpenClawConfig, ReplyToMode } from "openclaw/plugin-sdk/config-contracts"; import type { RuntimeEnv } from "openclaw/plugin-sdk/runtime-env"; import type { TelegramBotDeps } from "./bot-deps.js"; import type { TelegramBotInfo } from "./bot-info.js"; import type { TelegramTransport } from "./fetch.js"; +type DispatchReplyFromConfig = NonNullable; + export type TelegramBotOptions = { token: string; accountId?: string; @@ -12,6 +15,8 @@ export type TelegramBotOptions = { ownerAgentId?: string; runtime?: RuntimeEnv; buildContext?: typeof import("openclaw/plugin-sdk/channel-inbound").buildChannelInboundEventContext; + /** Instance-bound reply dispatcher prepared by the owning plugin runtime. */ + dispatchReplyFromConfig?: DispatchReplyFromConfig; requireMention?: boolean; allowFrom?: Array; groupAllowFrom?: Array; diff --git a/extensions/telegram/src/monitor.ts b/extensions/telegram/src/monitor.ts index 7281dbaae14b..54c7c1b0e5eb 100644 --- a/extensions/telegram/src/monitor.ts +++ b/extensions/telegram/src/monitor.ts @@ -154,6 +154,9 @@ export async function monitorTelegramProvider(opts: MonitorTelegramOpts = {}) { const proxyFetch = opts.proxyFetch ?? (account.config.proxy ? makeProxyFetch(account.config.proxy) : undefined); + // SAFETY: Gateway startup supplies the full plugin channel runtime; the surface type is the minimal external view. + const pluginChannelRuntime = opts.channelRuntime as PluginRuntime["channel"] | undefined; + if (opts.useWebhook) { const { startTelegramWebhook } = await loadTelegramMonitorWebhookRuntime(); if (isTelegramExecApprovalHandlerConfigured({ cfg, accountId: account.accountId })) { @@ -176,8 +179,9 @@ export async function monitorTelegramProvider(opts: MonitorTelegramOpts = {}) { secret: opts.webhookSecret ?? account.config.webhookSecret, host: opts.webhookHost ?? account.config.webhookHost, runtime: opts.runtime as RuntimeEnv, - buildContext: (opts.channelRuntime as PluginRuntime["channel"] | undefined)?.inbound - .buildContext, + buildContext: pluginChannelRuntime?.inbound.buildContext, + // Forward the owning runtime's bound dispatcher into the turn plan; never invoked here. + dispatchReplyFromConfig: pluginChannelRuntime?.reply?.dispatchReplyFromConfig, fetch: proxyFetch, abortSignal: opts.abortSignal, publicUrl: opts.webhookUrl, @@ -278,8 +282,9 @@ export async function monitorTelegramProvider(opts: MonitorTelegramOpts = {}) { accountId: account.accountId, ownerAgentId, runtime: opts.runtime, - buildContext: (opts.channelRuntime as PluginRuntime["channel"] | undefined)?.inbound - .buildContext, + buildContext: pluginChannelRuntime?.inbound.buildContext, + // Forward the owning runtime's bound dispatcher into the turn plan; never invoked here. + dispatchReplyFromConfig: pluginChannelRuntime?.reply?.dispatchReplyFromConfig, proxyFetch, botInfo: opts.botInfo, abortSignal: opts.abortSignal, diff --git a/extensions/telegram/src/polling-session.ts b/extensions/telegram/src/polling-session.ts index 9d03deb80c60..fdef864b9e18 100644 --- a/extensions/telegram/src/polling-session.ts +++ b/extensions/telegram/src/polling-session.ts @@ -89,6 +89,7 @@ type TelegramPollingSessionOpts = { ownerAgentId?: string; runtime: Parameters[0]["runtime"]; buildContext?: Parameters[0]["buildContext"]; + dispatchReplyFromConfig?: Parameters[0]["dispatchReplyFromConfig"]; proxyFetch: Parameters[0]["proxyFetch"]; botInfo?: Parameters[0]["botInfo"]; abortSignal?: AbortSignal; @@ -310,6 +311,7 @@ export class TelegramPollingSession { token: this.opts.token, runtime: this.opts.runtime, buildContext: this.opts.buildContext, + dispatchReplyFromConfig: this.opts.dispatchReplyFromConfig, proxyFetch: this.opts.proxyFetch, config: this.opts.config, accountId: this.opts.accountId, diff --git a/extensions/telegram/src/webhook.ts b/extensions/telegram/src/webhook.ts index 1fd9b4de7b27..455ccfbbf0c5 100644 --- a/extensions/telegram/src/webhook.ts +++ b/extensions/telegram/src/webhook.ts @@ -304,6 +304,7 @@ export async function startTelegramWebhook(opts: { secret?: string; runtime?: RuntimeEnv; buildContext?: Parameters[0]["buildContext"]; + dispatchReplyFromConfig?: Parameters[0]["dispatchReplyFromConfig"]; fetch?: typeof fetch; abortSignal?: AbortSignal; healthPath?: string; @@ -361,6 +362,7 @@ export async function startTelegramWebhook(opts: { token: opts.token, runtime, buildContext: opts.buildContext, + dispatchReplyFromConfig: opts.dispatchReplyFromConfig, proxyFetch: opts.fetch, fetchAbortSignal: botFetchAbortSignal, accountAbortSignal, diff --git a/extensions/whatsapp/src/auto-reply/monitor.ts b/extensions/whatsapp/src/auto-reply/monitor.ts index 40a8aec433e5..eddff91baef9 100644 --- a/extensions/whatsapp/src/auto-reply/monitor.ts +++ b/extensions/whatsapp/src/auto-reply/monitor.ts @@ -248,6 +248,10 @@ export async function monitorWebChannel( return meta?.participants?.length ? meta : undefined; }, createListener: async ({ sock, connection: connectionLocal }) => { + // SAFETY: Gateway startup supplies the full plugin channel runtime; the surface type is the minimal external view. + const pluginChannelRuntime = tuning.channelRuntime as + | PluginRuntime["channel"] + | undefined; const onMessage = createWebOnMessageHandler({ cfg, loadConfig: loadCurrentMonitorConfig, @@ -262,8 +266,9 @@ export async function monitorWebChannel( replyLogger, baseMentionConfig, account, - buildContext: (tuning.channelRuntime as PluginRuntime["channel"] | undefined)?.inbound - .buildContext, + buildContext: pluginChannelRuntime?.inbound.buildContext, + // Forward the owning runtime's bound dispatcher into the turn plan; never invoked here. + dispatchReplyFromConfig: pluginChannelRuntime?.reply?.dispatchReplyFromConfig, }); return (await (listenerFactory ?? attachWebInboxToSocket)({ cfg, diff --git a/extensions/whatsapp/src/auto-reply/monitor/on-message.ts b/extensions/whatsapp/src/auto-reply/monitor/on-message.ts index afed471b8617..67b45db16b27 100644 --- a/extensions/whatsapp/src/auto-reply/monitor/on-message.ts +++ b/extensions/whatsapp/src/auto-reply/monitor/on-message.ts @@ -1,5 +1,6 @@ // Whatsapp plugin module implements on message behavior. import type { AckReactionHandle } from "openclaw/plugin-sdk/channel-feedback"; +import type { ChannelInboundTurnPlan } from "openclaw/plugin-sdk/channel-inbound"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; import { ensureConfiguredBindingRouteReady, @@ -43,6 +44,7 @@ export function createWebOnMessageHandler(params: { baseMentionConfig: MentionConfig; account: { authDir?: string; accountId?: string; selfChatMode?: boolean }; buildContext?: typeof import("openclaw/plugin-sdk/channel-inbound").buildChannelInboundEventContext; + dispatchReplyFromConfig?: NonNullable; }) { const hasExplicitlyPassedInboundAccess = (msg: AdmittedWebInboundMessage): boolean => msg.admission.ingress.decision === "allow"; @@ -103,6 +105,7 @@ export function createWebOnMessageHandler(params: { replyLogger: params.replyLogger, backgroundTasks: params.backgroundTasks, buildContext: params.buildContext, + dispatchReplyFromConfig: params.dispatchReplyFromConfig, }; if (opts?.groupHistory !== undefined) { processParams.groupHistory = opts.groupHistory; diff --git a/extensions/whatsapp/src/auto-reply/monitor/process-message.test.ts b/extensions/whatsapp/src/auto-reply/monitor/process-message.test.ts index 98b89948b82e..1515337aa9cf 100644 --- a/extensions/whatsapp/src/auto-reply/monitor/process-message.test.ts +++ b/extensions/whatsapp/src/auto-reply/monitor/process-message.test.ts @@ -261,6 +261,7 @@ const baseRoute = { function callProcessMessage( overrides: { cfg?: unknown; + dispatchReplyFromConfig?: Parameters[0]["dispatchReplyFromConfig"]; groupHistories?: Map; msg?: unknown; } = {}, @@ -275,6 +276,7 @@ function callProcessMessage( connectionId: "conn-1", verbose: false, maxMediaBytes: 1024, + dispatchReplyFromConfig: overrides.dispatchReplyFromConfig, replyResolver: (async () => undefined) as never, replyLogger: { info: () => {}, warn: () => {}, error: () => {}, debug: () => {} } as never, backgroundTasks: new Set(), @@ -576,7 +578,7 @@ describe("processMessage group system prompt wiring", () => { ); }); - it("passes one lifecycle identity through the portable boundary and reply plan", async () => { + it("passes one lifecycle and owning dispatcher through the portable turn boundary", async () => { resolvePolicyMock.mockReturnValue(makePolicy(makeAccount())); buildContextMock.mockImplementationOnce(() => ({ Body: "hi", @@ -592,9 +594,13 @@ describe("processMessage group system prompt wiring", () => { onDeferred: vi.fn(), onAbandoned: vi.fn(async () => undefined), }; + const dispatchReplyFromConfig = vi.fn(async () => ({ + queuedFinal: false, + counts: { tool: 0, block: 0, final: 0 }, + })); const msg = attachWhatsAppIngressLifecycle(makeBaseMsg(), lifecycle as never); - await callProcessMessage({ msg }); + await callProcessMessage({ msg, dispatchReplyFromConfig }); const runParams = mockCallArg(runChannelInboundEventParamsMock, "runChannelInboundEvent") as { raw?: unknown; @@ -604,6 +610,7 @@ describe("processMessage group system prompt wiring", () => { turnAdoptionLifecycle?: unknown; }; expect(runParams.turnAdoptionLifecycle).toBe(replyPlanParams.turnAdoptionLifecycle); + expect(dispatchReplyFromConfig).toHaveBeenCalledOnce(); expect(runParams.raw).not.toHaveProperty("platform"); expect(runParams.raw).not.toHaveProperty("admission"); }); diff --git a/extensions/whatsapp/src/auto-reply/monitor/process-message.ts b/extensions/whatsapp/src/auto-reply/monitor/process-message.ts index 8c5393ce5841..8bfdf0e30187 100644 --- a/extensions/whatsapp/src/auto-reply/monitor/process-message.ts +++ b/extensions/whatsapp/src/auto-reply/monitor/process-message.ts @@ -6,6 +6,7 @@ import { } from "openclaw/plugin-sdk/channel-feedback"; import { type buildChannelInboundEventContext, + type ChannelInboundTurnPlan, formatMediaPlaceholderText, runChannelInboundEvent, } from "openclaw/plugin-sdk/channel-inbound"; @@ -214,6 +215,7 @@ export async function processMessage(params: { * - undefined (omitted) → caller did not attempt preflight; run internal STT as normal */ preflightAudioTranscript?: string | null; buildContext?: typeof buildChannelInboundEventContext; + dispatchReplyFromConfig?: NonNullable; }) { const admission = requireWhatsAppInboundAdmission(params.msg); if (admission.ingress.admission !== "dispatch" && admission.ingress.admission !== "observe") { @@ -586,6 +588,7 @@ export async function processMessage(params: { accountId: params.route.accountId, route: { agentId: params.route.agentId, sessionKey: params.route.sessionKey }, ctxPayload, + dispatchReplyFromConfig: params.dispatchReplyFromConfig, record: { onRecordError: (err) => { params.replyLogger.warn( diff --git a/src/auto-reply/reply/provider-dispatcher.test.ts b/src/auto-reply/reply/provider-dispatcher.test.ts index 55c931b9e695..b46794986ce6 100644 --- a/src/auto-reply/reply/provider-dispatcher.test.ts +++ b/src/auto-reply/reply/provider-dispatcher.test.ts @@ -1,5 +1,6 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import type { OpenClawConfig } from "../../config/types.openclaw.js"; +import type { DispatchReplyFromConfig } from "./dispatch-from-config.types.js"; import type { ReplyDispatcherOptions, ReplyDispatcherWithTypingOptions, @@ -36,16 +37,18 @@ describe("provider dispatcher wrappers", () => { hoisted.plainDispatchMock.mockResolvedValue(dispatchResult); }); - it("forwards runtime toolsAllow through the buffered wrapper", async () => { + it("forwards allowed tools and the owning dispatcher through the buffered wrapper", async () => { const dispatcherOptions = { deliver: async () => ({ visibleReplySent: false }), } satisfies ReplyDispatcherWithTypingOptions; + const dispatchReplyFromConfig = vi.fn(); await dispatchReplyWithBufferedBlockDispatcherCore({ ctx: { Body: "hello" }, cfg: {} as OpenClawConfig, dispatcherOptions, toolsAllow: ["message"], + dispatchReplyFromConfig, }); expect(hoisted.bufferedDispatchMock).toHaveBeenCalledTimes(1); @@ -53,6 +56,7 @@ describe("provider dispatcher wrappers", () => { expect.objectContaining({ dispatcherOptions, toolsAllow: ["message"], + dispatchReplyFromConfig, }), ); }); diff --git a/src/auto-reply/reply/provider-dispatcher.ts b/src/auto-reply/reply/provider-dispatcher.ts index 519d85899417..8f4e33632845 100644 --- a/src/auto-reply/reply/provider-dispatcher.ts +++ b/src/auto-reply/reply/provider-dispatcher.ts @@ -23,6 +23,7 @@ export const dispatchReplyWithBufferedBlockDispatcherCore: DispatchReplyWithBuff toolsAllow: params.toolsAllow, replyResolver: params.replyResolver, replyOptions: params.replyOptions, + dispatchReplyFromConfig: params.dispatchReplyFromConfig, }); }; diff --git a/src/auto-reply/reply/provider-dispatcher.types.ts b/src/auto-reply/reply/provider-dispatcher.types.ts index 65dc61df0522..9bc4f77382b0 100644 --- a/src/auto-reply/reply/provider-dispatcher.types.ts +++ b/src/auto-reply/reply/provider-dispatcher.types.ts @@ -3,7 +3,10 @@ import type { OpenClawConfig } from "../../config/types.openclaw.js"; import type { PluginCommandReplyOptions } from "../../plugins/plugin-command-dispatch-contract.js"; import type { GetReplyOptions } from "../get-reply-options.types.js"; import type { FinalizedMsgContext, MsgContext } from "../templating.js"; -import type { DispatchFromConfigResult } from "./dispatch-from-config.types.js"; +import type { + DispatchFromConfigResult, + DispatchReplyFromConfig, +} from "./dispatch-from-config.types.js"; import type { GetReplyFromConfig } from "./get-reply.types.js"; import type { ReplyDispatcherOptions, @@ -21,6 +24,7 @@ export type DispatchReplyWithBufferedBlockDispatcher = (params: { toolsAllow?: string[]; replyOptions?: DispatchReplyOptions; replyResolver?: GetReplyFromConfig; + dispatchReplyFromConfig?: DispatchReplyFromConfig; }) => Promise; /** Plain dispatcher entry point used when block buffering is not needed. */ diff --git a/src/channels/message-access/context-builder-callers.test.ts b/src/channels/message-access/context-builder-callers.test.ts index 7b0a11e49f18..0b4ce0c802e3 100644 --- a/src/channels/message-access/context-builder-callers.test.ts +++ b/src/channels/message-access/context-builder-callers.test.ts @@ -153,12 +153,12 @@ const SCOPED_BUILDER_HANDOFFS = [ [ "discord", "extensions/discord/src/monitor/provider.ts", - 'opts.channelRuntime as PluginRuntime["channel"] | undefined', + "buildContext: pluginChannelRuntime?.inbound.buildContext", ], [ "imessage", "extensions/imessage/src/monitor/monitor-provider.ts", - 'opts.channelRuntime as PluginRuntime["channel"] | undefined', + "buildContext: pluginChannelRuntime?.inbound.buildContext", ], [ "line", @@ -181,7 +181,7 @@ const SCOPED_BUILDER_HANDOFFS = [ [ "telegram-webhook", "extensions/telegram/src/monitor.ts", - 'opts.channelRuntime as PluginRuntime["channel"] | undefined', + "buildContext: pluginChannelRuntime?.inbound.buildContext", ], ["telegram-webhook", "extensions/telegram/src/webhook.ts", "buildContext: opts.buildContext"], [ @@ -198,7 +198,7 @@ const SCOPED_BUILDER_HANDOFFS = [ [ "whatsapp", "extensions/whatsapp/src/auto-reply/monitor.ts", - 'tuning.channelRuntime as PluginRuntime["channel"] | undefined', + "buildContext: pluginChannelRuntime?.inbound.buildContext", ], [ "whatsapp", @@ -236,12 +236,6 @@ describe("channel context builder caller inventory", () => { expect(source("extensions/signal/src/monitor.ts")).toContain( "channelRuntime: opts.channelRuntime", ); - expect(source("extensions/telegram/src/monitor.ts")).toContain( - 'opts.channelRuntime as PluginRuntime["channel"] | undefined', - ); - expect(source("extensions/whatsapp/src/auto-reply/monitor.ts")).toContain( - 'tuning.channelRuntime as PluginRuntime["channel"] | undefined', - ); }); it("does not rediscover host context builders through process-global runtime stores", () => { diff --git a/src/gateway/server-plugins.test.ts b/src/gateway/server-plugins.test.ts index c56deb43088e..c6af816443f9 100644 --- a/src/gateway/server-plugins.test.ts +++ b/src/gateway/server-plugins.test.ts @@ -5,6 +5,7 @@ import path from "node:path"; import { isRecord } from "@openclaw/normalization-core/record-coerce"; import { createRequireRecord } from "openclaw/plugin-sdk/test-fixtures"; import { afterEach, beforeAll, beforeEach, describe, expect, test, vi } from "vitest"; +import { createTerminalTool } from "../agents/tools/terminal-tool.js"; import { getGlobalPluginRegistry, initializeGlobalHookRunner, @@ -596,12 +597,29 @@ describe("loadGatewayPlugins", () => { }); test("binds channel reply dispatch to the owning Gateway context", async () => { + const terminalSessions = { listAgent: vi.fn(() => []) }; const context = { label: "channel-reply-owner", workerSessionPlacementService: { getMany: vi.fn(() => new Map()) }, + terminalSessions, } as unknown as GatewayRequestContext; serverPluginsModule.setFallbackGatewayContext(context); loadOpenClawPlugins.mockReturnValue(createRegistry([])); + dispatchReplyFromConfig.mockImplementationOnce(async () => { + const result = await createTerminalTool({ + agentId: "main", + agentSessionKey: "agent:main:telegram:direct:123", + sessionId: "session-123", + }).execute("terminal-list", { action: "list" }); + expect(result.details).toEqual({ sessions: [] }); + expect(terminalSessions.listAgent).toHaveBeenCalledWith({ + kind: "agent", + agentId: "main", + agentSessionKey: "agent:main:telegram:direct:123", + agentSessionId: "session-123", + }); + return { counts: {}, queuedFinal: false }; + }); loadGatewayPluginsForTest(); const runtimeOptions = getLastPluginLoadOption("runtimeOptions") as