Skip to content

Commit 6a0fdea

Browse files
vincentkocvincentkoc
committed
fix(outbound): materialize buffer-only sends
Fixes #90768 Incorporates the send-buffer materialization shape proposed in #90794 by @LiuwqGit, with maintainer fixes for dry-run, gateway delivery, byte-cap, target-validation, and downstream plugin dispatch paths.
1 parent 85840eb commit 6a0fdea

16 files changed

Lines changed: 939 additions & 47 deletions

‎packages/gateway-protocol/src/schema/agent.ts‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -126,6 +126,12 @@ export const SendParamsSchema = Type.Object(
126126
message: Type.Optional(Type.String()),
127127
mediaUrl: Type.Optional(Type.String()),
128128
mediaUrls: Type.Optional(Type.Array(Type.String())),
129+
/** Base64 attachment payload for gateway-local media materialization. */
130+
buffer: Type.Optional(Type.String()),
131+
/** Optional filename for a base64 attachment payload. */
132+
filename: Type.Optional(Type.String()),
133+
/** Optional MIME type for a base64 attachment payload. */
134+
contentType: Type.Optional(Type.String()),
129135
asVoice: Type.Optional(Type.Boolean()),
130136
gifPlayback: Type.Optional(Type.Boolean()),
131137
channel: Type.Optional(Type.String()),

‎packages/media-core/src/media-source-url.ts‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,13 @@
11
// Media Core module implements media source url behavior.
22
const HTTP_URL_RE = /^https?:\/\//i;
33
const MXC_URL_RE = /^mxc:\/\//i;
4+
const BUFFER_URL_RE = /^buffer:\/\//i;
45

56
/** Returns true for remote media URLs that should stay URL-backed instead of local-file-backed. */
67
export function isPassThroughRemoteMediaSource(value: string | null | undefined): boolean {
78
const normalized = value?.trim() ?? "";
8-
return Boolean(normalized) && (HTTP_URL_RE.test(normalized) || MXC_URL_RE.test(normalized));
9+
return (
10+
Boolean(normalized) &&
11+
(HTTP_URL_RE.test(normalized) || MXC_URL_RE.test(normalized) || BUFFER_URL_RE.test(normalized))
12+
);
913
}

‎src/gateway/server-methods/send.test.ts‎

Lines changed: 108 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,8 @@
11
// Send method tests cover outbound message routing, transcript mirroring, poll
22
// dispatch, plugin channel selection, and durable delivery dependencies.
3+
import fs from "node:fs/promises";
4+
import os from "node:os";
5+
import path from "node:path";
36
import { beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
47
import { jsonResult } from "../../agents/tools/common.js";
58
import type { ChannelPlugin } from "../../channels/plugins/types.js";
@@ -209,6 +212,22 @@ async function runMessageActionRequest(
209212
return { respond };
210213
}
211214

215+
async function withTempOpenClawStateDir<T>(test: (stateDir: string) => Promise<T>): Promise<T> {
216+
const previous = process.env.OPENCLAW_STATE_DIR;
217+
const stateDir = await fs.mkdtemp(path.join(os.tmpdir(), "gateway-send-state-"));
218+
process.env.OPENCLAW_STATE_DIR = stateDir;
219+
try {
220+
return await test(stateDir);
221+
} finally {
222+
if (previous === undefined) {
223+
delete process.env.OPENCLAW_STATE_DIR;
224+
} else {
225+
process.env.OPENCLAW_STATE_DIR = previous;
226+
}
227+
await fs.rm(stateDir, { recursive: true, force: true });
228+
}
229+
}
230+
212231
function deliveryCall(index = 0): Record<string, any> | undefined {
213232
const calls = mocks.deliverOutboundPayloads.mock.calls as unknown as Array<[Record<string, any>]>;
214233
return calls[index]?.[0];
@@ -792,6 +811,34 @@ describe("gateway send mirroring", () => {
792811
expect(deliveryCall()?.session?.key).toBe("agent:work:whatsapp:resolved");
793812
});
794813

814+
it("materializes buffer-only gateway sends before outbound delivery", async () => {
815+
mockDeliverySuccess("m-buffer-media");
816+
817+
await withTempOpenClawStateDir(async () => {
818+
const { respond } = await runSend({
819+
to: "+15551234567",
820+
mediaUrl: "buffer://message-send/attachment",
821+
mediaUrls: ["buffer://message-send/attachment"],
822+
buffer: Buffer.from("gateway send bytes").toString("base64"),
823+
filename: "gateway-send.txt",
824+
contentType: "text/plain",
825+
channel: "whatsapp",
826+
agentId: "work",
827+
idempotencyKey: "idem-whatsapp-buffer",
828+
});
829+
830+
expect(firstRespondCall(respond)[0]).toBe(true);
831+
const payload = deliveryCall()?.payloads?.[0];
832+
expect(typeof payload?.mediaUrl).toBe("string");
833+
expect(payload?.mediaUrls).toEqual([payload?.mediaUrl]);
834+
expect(payload?.mediaUrl).not.toBe("buffer://message-send/attachment");
835+
await expect(fs.readFile(String(payload?.mediaUrl), "utf8")).resolves.toBe(
836+
"gateway send bytes",
837+
);
838+
expect(deliveryCall()?.session?.agentId).toBe("work");
839+
});
840+
});
841+
795842
it("maps gateway asVoice sends onto outbound audioAsVoice payloads", async () => {
796843
mockDeliverySuccess("m-voice");
797844

@@ -2243,4 +2290,65 @@ describe("gateway send mirroring", () => {
22432290
expect(actionCall?.mediaLocalRoots).toContain(TEST_AGENT_WORKSPACE);
22442291
expect(actionCall?.gatewayClientScopes).toEqual(["operator.write"]);
22452292
});
2293+
2294+
it("materializes buffer-only message.action sends on the gateway before plugin dispatch", async () => {
2295+
const mediaActionPlugin: ChannelPlugin = {
2296+
id: "telegram",
2297+
meta: {
2298+
id: "telegram",
2299+
label: "Telegram",
2300+
selectionLabel: "Telegram",
2301+
docsPath: "/channels/telegram",
2302+
blurb: "Telegram media action dispatch test plugin.",
2303+
},
2304+
capabilities: { chatTypes: ["direct"] },
2305+
config: {
2306+
listAccountIds: () => ["default"],
2307+
resolveAccount: () => ({ enabled: true }),
2308+
isConfigured: () => true,
2309+
},
2310+
actions: {
2311+
describeMessageTool: () => ({ actions: ["send"] }),
2312+
supportsAction: ({ action }) => action === "send",
2313+
handleAction: async () => jsonResult({ ok: true }),
2314+
},
2315+
};
2316+
mocks.getChannelPlugin.mockReturnValue(mediaActionPlugin);
2317+
setActivePluginRegistry(
2318+
createTestRegistry([{ pluginId: "telegram", source: "test", plugin: mediaActionPlugin }]),
2319+
"send-test-message-action-buffer-materialize",
2320+
);
2321+
2322+
await withTempOpenClawStateDir(async () => {
2323+
const { respond } = await runMessageActionRequest(
2324+
{
2325+
channel: "telegram",
2326+
action: "send",
2327+
params: {
2328+
to: "123",
2329+
media: "buffer://message-send/attachment",
2330+
mediaUrl: "buffer://message-send/attachment",
2331+
mediaUrls: ["buffer://message-send/attachment"],
2332+
buffer: Buffer.from("gateway bytes").toString("base64"),
2333+
filename: "gateway.txt",
2334+
contentType: "text/plain",
2335+
},
2336+
agentId: "work",
2337+
idempotencyKey: "idem-message-action-buffer-materialize",
2338+
},
2339+
{ connect: { scopes: ["operator.write"] } },
2340+
);
2341+
2342+
expect(firstRespondCall(respond)[0]).toBe(true);
2343+
const actionCall = lastDispatchChannelMessageActionCall();
2344+
const actionParams = actionCall?.params;
2345+
expect(actionParams?.buffer).toBeUndefined();
2346+
expect(typeof actionParams?.mediaUrl).toBe("string");
2347+
expect(actionParams?.media).toBe(actionParams?.mediaUrl);
2348+
expect(actionParams?.mediaUrls).toEqual([actionParams?.mediaUrl]);
2349+
await expect(fs.readFile(String(actionParams?.mediaUrl), "utf8")).resolves.toBe(
2350+
"gateway bytes",
2351+
);
2352+
});
2353+
});
22462354
});

‎src/gateway/server-methods/send.ts‎

Lines changed: 66 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,10 @@ import {
2626
import type { OpenClawConfig } from "../../config/types.openclaw.js";
2727
import { resolveOutboundChannelPlugin } from "../../infra/outbound/channel-resolution.js";
2828
import { resolveMessageChannelSelection } from "../../infra/outbound/channel-selection.js";
29+
import {
30+
hydrateAttachmentParamsForAction,
31+
resolveAttachmentMediaPolicy,
32+
} from "../../infra/outbound/message-action-params.js";
2933
import {
3034
ensureOutboundSessionEntry,
3135
resolveOutboundSessionRoute,
@@ -507,25 +511,39 @@ export const sendHandlers: GatewayRequestHandlers = {
507511
}
508512

509513
try {
514+
const sessionKey = normalizeOptionalString(request.sessionKey) ?? undefined;
515+
const agentId =
516+
normalizeOptionalString(request.agentId) ??
517+
(sessionKey ? resolveSessionAgentId({ sessionKey, config: cfg }) : undefined);
518+
const accountId = normalizeOptionalString(request.accountId) ?? undefined;
519+
if (request.action === "send") {
520+
await hydrateAttachmentParamsForAction({
521+
cfg,
522+
channel,
523+
accountId,
524+
args: request.params,
525+
action: "send",
526+
mediaPolicy: resolveAttachmentMediaPolicy({
527+
mediaLocalRoots: getAgentScopedMediaLocalRoots(cfg, agentId),
528+
}),
529+
});
530+
}
510531
const gatewayClientScopes = client?.connect?.scopes ?? [];
511532
const handled = await dispatchChannelMessageAction({
512533
channel,
513534
action: request.action as never,
514535
cfg,
515536
params: request.params,
516-
accountId: normalizeOptionalString(request.accountId) ?? undefined,
537+
accountId,
517538
requesterSenderId: normalizeOptionalString(request.requesterSenderId) ?? undefined,
518539
senderIsOwner: gatewayClientScopes.includes(ADMIN_SCOPE)
519540
? request.senderIsOwner === true
520541
: false,
521-
sessionKey: normalizeOptionalString(request.sessionKey) ?? undefined,
542+
sessionKey,
522543
sessionId: normalizeOptionalString(request.sessionId) ?? undefined,
523544
inboundEventKind: request.inboundTurnKind,
524-
agentId: normalizeOptionalString(request.agentId) ?? undefined,
525-
mediaLocalRoots: getAgentScopedMediaLocalRoots(
526-
cfg,
527-
normalizeOptionalString(request.agentId) ?? undefined,
528-
),
545+
agentId,
546+
mediaLocalRoots: getAgentScopedMediaLocalRoots(cfg, agentId),
529547
toolContext: request.toolContext,
530548
dryRun: false,
531549
gatewayClientScopes,
@@ -539,10 +557,6 @@ export const sendHandlers: GatewayRequestHandlers = {
539557
return { ok: false, error, meta: { channel } };
540558
}
541559
const payload = extractToolPayload(handled);
542-
const sessionKey = normalizeOptionalString(request.sessionKey) ?? undefined;
543-
const agentId =
544-
normalizeOptionalString(request.agentId) ??
545-
(sessionKey ? resolveSessionAgentId({ sessionKey, config: cfg }) : undefined);
546560
await scheduleDeliveredSourceReplyTranscriptMirror({
547561
context,
548562
mirror: {
@@ -583,6 +597,9 @@ export const sendHandlers: GatewayRequestHandlers = {
583597
message?: string;
584598
mediaUrl?: string;
585599
mediaUrls?: string[];
600+
buffer?: string;
601+
filename?: string;
602+
contentType?: string;
586603
asVoice?: boolean;
587604
gifPlayback?: boolean;
588605
channel?: string;
@@ -615,7 +632,8 @@ export const sendHandlers: GatewayRequestHandlers = {
615632
.map((entry) => normalizeOptionalString(entry))
616633
.filter((entry): entry is string => Boolean(entry))
617634
: undefined;
618-
if (!message && !mediaUrl && (mediaUrls?.length ?? 0) === 0) {
635+
const buffer = readStringValue(request.buffer);
636+
if (!message && !mediaUrl && (mediaUrls?.length ?? 0) === 0 && !buffer) {
619637
respond(
620638
false,
621639
undefined,
@@ -663,19 +681,6 @@ export const sendHandlers: GatewayRequestHandlers = {
663681
accountId,
664682
});
665683
const deliveryTarget = idLikeTarget?.to ?? resolvedTarget.to;
666-
const outboundDeps = context.deps ? createOutboundSendDeps(context.deps) : undefined;
667-
const outboundPayloads = [
668-
{
669-
text: message,
670-
mediaUrl,
671-
mediaUrls,
672-
...(request.asVoice === true ? { audioAsVoice: true } : {}),
673-
},
674-
];
675-
const outboundPayloadPlan = createOutboundPayloadPlan(outboundPayloads);
676-
const mirrorProjection = projectOutboundPayloadPlanForMirror(outboundPayloadPlan);
677-
const mirrorText = mirrorProjection.text;
678-
const mirrorMediaUrls = mirrorProjection.mediaUrls;
679684
// Preserve opaque, case-sensitive peer IDs (e.g. Matrix room ids) on an
680685
// explicit session key instead of raw-lowercasing it (openclaw#75670).
681686
// Non-enrolled channels still canonicalize to lowercase via the registry.
@@ -687,6 +692,42 @@ export const sendHandlers: GatewayRequestHandlers = {
687692
: undefined;
688693
const defaultAgentId = resolveSessionAgentId({ config: cfg });
689694
const effectiveAgentId = explicitAgentId ?? sessionAgentId ?? defaultAgentId;
695+
const sendArgs: Record<string, unknown> = {
696+
mediaUrl,
697+
mediaUrls,
698+
buffer,
699+
filename: normalizeOptionalString(request.filename) ?? undefined,
700+
contentType: normalizeOptionalString(request.contentType) ?? undefined,
701+
};
702+
await hydrateAttachmentParamsForAction({
703+
cfg,
704+
channel,
705+
accountId,
706+
args: sendArgs,
707+
action: "send",
708+
mediaPolicy: resolveAttachmentMediaPolicy({
709+
mediaLocalRoots: getAgentScopedMediaLocalRoots(cfg, effectiveAgentId),
710+
}),
711+
});
712+
const hydratedMediaUrl = normalizeOptionalString(sendArgs.mediaUrl);
713+
const hydratedMediaUrls = Array.isArray(sendArgs.mediaUrls)
714+
? sendArgs.mediaUrls
715+
.map((entry) => normalizeOptionalString(entry))
716+
.filter((entry): entry is string => Boolean(entry))
717+
: undefined;
718+
const outboundDeps = context.deps ? createOutboundSendDeps(context.deps) : undefined;
719+
const outboundPayloads = [
720+
{
721+
text: message,
722+
mediaUrl: hydratedMediaUrl,
723+
mediaUrls: hydratedMediaUrls,
724+
...(request.asVoice === true ? { audioAsVoice: true } : {}),
725+
},
726+
];
727+
const outboundPayloadPlan = createOutboundPayloadPlan(outboundPayloads);
728+
const mirrorProjection = projectOutboundPayloadPlanForMirror(outboundPayloadPlan);
729+
const mirrorText = mirrorProjection.text;
730+
const mirrorMediaUrls = mirrorProjection.mediaUrls;
690731
const derivedRoute = await resolveOutboundSessionRoute({
691732
cfg,
692733
channel,

‎src/infra/outbound/message-action-param-keys.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ const STANDARD_MESSAGE_ACTION_PARAM_KEYS = new Set([
88
"attachments",
99
"base64",
1010
"bestEffort",
11+
"buffer",
1112
"caption",
1213
"channel",
1314
"channelId",

0 commit comments

Comments
 (0)