vultr/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-05-06 17:40:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: migrate bundled plugins to message lifecycle

Browse Source
This commit is contained in:
Peter Steinberger
2026-05-06 01:40:53 +01:00
parent 2ead1502c9
commit 05eda57b3c
223 changed files with 8568 additions and 1354 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
extensions/matrix/src/approval-handler.runtime.test.ts
View File

@@ -21,6 +21,29 @@ type MatrixPendingPluginApprovalView = Extract<
const MATRIX_APPROVAL_METADATA_KEY = "com.openclaw.approval";
function buildMatrixReceipt(messageIds: readonly string[], roomId = "!room:example.org") {
return {
primaryPlatformMessageId: messageIds[0],
platformMessageIds: [...messageIds],
parts: messageIds.map((messageId, index) => ({
platformMessageId: messageId,
kind: "text" as const,
index,
raw: {
channel: "matrix",
messageId,
roomId,
},
})),
sentAt: 100,
raw: messageIds.map((messageId) => ({
channel: "matrix",
messageId,
roomId,
})),
};
}
function buildMatrixApprovalRoomTarget(
roomId: string,
): MatrixDeliverPendingParams["plannedTarget"] {
@@ -142,7 +165,7 @@ describe("matrixApprovalNativeRuntime", () => {
const sendSingleTextMessage = vi.fn().mockResolvedValue({
messageId: "$approval",
primaryMessageId: "$approval",
messageIds: ["$approval"],
receipt: buildMatrixReceipt(["$approval"]),
roomId: "!room:example.org",
});
const reactMessage = vi.fn().mockResolvedValue(undefined);
@@ -195,7 +218,7 @@ describe("matrixApprovalNativeRuntime", () => {
const sendSingleTextMessage = vi.fn().mockResolvedValue({
messageId: "$plugin-approval",
primaryMessageId: "$plugin-approval",
messageIds: ["$plugin-approval"],
receipt: buildMatrixReceipt(["$plugin-approval"]),
roomId: "!room:example.org",
});
const reactMessage = vi.fn().mockResolvedValue(undefined);
@@ -270,7 +293,7 @@ describe("matrixApprovalNativeRuntime", () => {
const sendSingleTextMessage = vi.fn().mockResolvedValue({
messageId: "$approval",
primaryMessageId: "$approval",
messageIds: ["$approval"],
receipt: buildMatrixReceipt(["$approval"]),
roomId: "!room:example.org",
});
const reactMessage = vi.fn().mockImplementation(async () => {
@@ -318,8 +341,8 @@ describe("matrixApprovalNativeRuntime", () => {
.mockRejectedValue(new Error("Matrix single-message text exceeds limit (5000 > 4000)"));
const sendMessage = vi.fn().mockResolvedValue({
messageId: "$last",
primaryMessageId: "$primary",
messageIds: ["$primary", "$last"],
primaryMessageId: "$legacy-primary",
receipt: buildMatrixReceipt(["$primary", "$last"]),
roomId: "!room:example.org",
});
const reactMessage = vi.fn().mockResolvedValue(undefined);
@@ -375,7 +398,7 @@ describe("matrixApprovalNativeRuntime", () => {
);
expect(entry).toMatchObject({
roomId: "!room:example.org",
messageIds: ["$primary", "$last"],
platformMessageIds: ["$primary", "$last"],
reactionEventId: "$primary",
});
const bindPending = matrixApprovalNativeRuntime.interactions?.bindPending;

28
extensions/matrix/src/approval-handler.runtime.ts
View File

@@ -15,6 +15,10 @@ import type {
ExecApprovalRequest,
PluginApprovalRequest,
} from "openclaw/plugin-sdk/approval-runtime";
import {
listMessageReceiptPlatformIds,
resolveMessageReceiptPrimaryId,
} from "openclaw/plugin-sdk/channel-message";
import {
buildMatrixApprovalReactionHint,
listMatrixApprovalReactionBindings,
@@ -42,7 +46,7 @@ const MATRIX_APPROVAL_METADATA_KEY = "com.openclaw.approval" as const;
type PendingMessage = {
roomId: string;
messageIds: readonly string[];
platformMessageIds: readonly string[];
reactionEventId: string;
};
type PreparedMatrixTarget = {
@@ -147,7 +151,9 @@ function resolveHandlerContext(params: ChannelApprovalCapabilityHandlerContext):
}
function normalizePendingMessageIds(entry: PendingMessage): string[] {
return Array.from(new Set(entry.messageIds.map((messageId) => messageId.trim()).filter(Boolean)));
return Array.from(
new Set(entry.platformMessageIds.map((messageId) => messageId.trim()).filter(Boolean)),
);
}
function normalizeReactionTargetRef(params: ReactionTargetRef): ReactionTargetRef | null {
@@ -438,15 +444,15 @@ export const matrixApprovalNativeRuntime = createChannelApprovalNativeRuntimeAda
extraContent: pendingPayload.extraContent,
});
}
const messageIds = Array.from(
new Set(
(result.messageIds ?? [result.messageId])
.map((messageId) => messageId.trim())
.filter(Boolean),
),
);
const receiptMessageIds = listMessageReceiptPlatformIds(result.receipt);
const platformMessageIds = receiptMessageIds.length
? receiptMessageIds
: [result.messageId.trim()].filter(Boolean);
const reactionEventId =
result.primaryMessageId?.trim() || messageIds[0] || result.messageId.trim();
resolveMessageReceiptPrimaryId(result.receipt) ||
result.primaryMessageId?.trim() ||
platformMessageIds[0] ||
result.messageId.trim();
registerMatrixApprovalReactionTarget({
roomId: result.roomId,
eventId: reactionEventId,
@@ -467,7 +473,7 @@ export const matrixApprovalNativeRuntime = createChannelApprovalNativeRuntimeAda
);
return {
roomId: result.roomId,
messageIds,
platformMessageIds,
reactionEventId,
};
},

169
extensions/matrix/src/channel.message-adapter.test.ts Normal file
View File

@@ -0,0 +1,169 @@
import {
verifyChannelMessageAdapterCapabilityProofs,
verifyChannelMessageLiveCapabilityAdapterProofs,
verifyChannelMessageLiveFinalizerProofs,
} from "openclaw/plugin-sdk/channel-message";
import { beforeEach, describe, expect, it, vi } from "vitest";
import type { OpenClawConfig } from "../runtime-api.js";
const mocks = vi.hoisted(() => ({
sendMessageMatrix: vi.fn(),
}));
vi.mock("./matrix/send.js", () => ({
sendMessageMatrix: mocks.sendMessageMatrix,
sendPollMatrix: vi.fn(),
sendTypingMatrix: vi.fn(),
}));
vi.mock("./runtime.js", () => ({
getMatrixRuntime: () => ({
channel: {
text: {
chunkMarkdownText: (text: string) => [text],
},
},
}),
}));
import { matrixPlugin } from "./channel.js";
const cfg = {
channels: {
matrix: {
accessToken: "resolved-token",
},
},
} as OpenClawConfig;
describe("matrix channel message adapter", () => {
beforeEach(() => {
mocks.sendMessageMatrix.mockReset();
mocks.sendMessageMatrix.mockResolvedValue({ messageId: "$event-1", roomId: "!room:example" });
});
it("backs declared durable-final capabilities with runtime outbound proofs", async () => {
const adapter = matrixPlugin.message;
expect(adapter).toBeDefined();
const proveText = async () => {
mocks.sendMessageMatrix.mockClear();
const result = await adapter!.send!.text!({
cfg,
to: "room:!room:example",
text: "hello",
accountId: "default",
});
expect(mocks.sendMessageMatrix).toHaveBeenLastCalledWith(
"room:!room:example",
"hello",
expect.objectContaining({ cfg, accountId: "default" }),
);
expect(result.receipt.platformMessageIds).toEqual(["$event-1"]);
expect(result.receipt.parts[0]?.kind).toBe("text");
};
const proveMedia = async () => {
mocks.sendMessageMatrix.mockClear();
const result = await adapter!.send!.media!({
cfg,
to: "room:!room:example",
text: "caption",
mediaUrl: "file:///tmp/cat.png",
mediaLocalRoots: ["/tmp/openclaw"],
accountId: "default",
audioAsVoice: true,
});
expect(mocks.sendMessageMatrix).toHaveBeenLastCalledWith(
"room:!room:example",
"caption",
expect.objectContaining({
cfg,
mediaUrl: "file:///tmp/cat.png",
mediaLocalRoots: ["/tmp/openclaw"],
audioAsVoice: true,
}),
);
expect(result.receipt.parts[0]?.kind).toBe("voice");
};
const proveReplyThread = async () => {
mocks.sendMessageMatrix.mockClear();
const result = await adapter!.send!.text!({
cfg,
to: "room:!room:example",
text: "threaded",
accountId: "default",
replyToId: "$reply",
threadId: "$thread",
});
expect(mocks.sendMessageMatrix).toHaveBeenLastCalledWith(
"room:!room:example",
"threaded",
expect.objectContaining({
cfg,
replyToId: "$reply",
threadId: "$thread",
}),
);
expect(result.receipt.replyToId).toBe("$reply");
expect(result.receipt.threadId).toBe("$thread");
};
await verifyChannelMessageAdapterCapabilityProofs({
adapterName: "matrixMessageAdapter",
adapter: adapter!,
proofs: {
text: proveText,
media: proveMedia,
replyTo: proveReplyThread,
thread: proveReplyThread,
messageSendingHooks: () => {
expect(adapter!.send!.text).toBeTypeOf("function");
},
},
});
});
it("backs declared live preview finalizer capabilities with adapter proofs", async () => {
const adapter = matrixPlugin.message;
await verifyChannelMessageLiveCapabilityAdapterProofs({
adapterName: "matrixMessageAdapter",
adapter: adapter!,
proofs: {
draftPreview: () => {
expect(adapter!.live?.finalizer?.capabilities?.discardPending).toBe(true);
},
previewFinalization: () => {
expect(adapter!.live?.finalizer?.capabilities?.finalEdit).toBe(true);
},
progressUpdates: () => {
expect(adapter!.live?.capabilities?.draftPreview).toBe(true);
},
quietFinalization: () => {
expect(adapter!.live?.finalizer?.capabilities?.previewReceipt).toBe(true);
},
},
});
await verifyChannelMessageLiveFinalizerProofs({
adapterName: "matrixMessageAdapter",
adapter: adapter!,
proofs: {
finalEdit: () => {
expect(adapter!.live?.capabilities?.previewFinalization).toBe(true);
},
normalFallback: () => {
expect(adapter!.send!.text).toBeTypeOf("function");
},
discardPending: () => {
expect(adapter!.live?.capabilities?.draftPreview).toBe(true);
},
previewReceipt: () => {
expect(adapter!.live?.capabilities?.quietFinalization).toBe(true);
},
},
});
});
});

89
extensions/matrix/src/channel.ts
View File

@@ -5,10 +5,12 @@ import {
} from "openclaw/plugin-sdk/channel-config-helpers";
import type { ChannelDoctorAdapter } from "openclaw/plugin-sdk/channel-contract";
import { createChatChannelPlugin, type ChannelPlugin } from "openclaw/plugin-sdk/channel-core";
import { createChannelMessageAdapterFromOutbound } from "openclaw/plugin-sdk/channel-message";
import {
createAllowlistProviderOpenWarningCollector,
projectAccountConfigWarningCollector,
} from "openclaw/plugin-sdk/channel-policy";
import type { ChannelOutboundAdapter } from "openclaw/plugin-sdk/channel-send-result";
import { createScopedAccountReplyToModeResolver } from "openclaw/plugin-sdk/conversation-runtime";
import {
createChannelDirectoryAdapter,
@@ -319,6 +321,64 @@ function resolveMatrixDeliveryTarget(params: {
return null;
}
const matrixChannelOutbound: ChannelOutboundAdapter = {
deliveryMode: "direct",
chunker: chunkTextForOutbound,
chunkerMode: "markdown",
textChunkLimit: 4000,
deliveryCapabilities: {
durableFinal: {
text: true,
media: true,
replyTo: true,
thread: true,
messageSendingHooks: true,
},
},
shouldSuppressLocalPayloadPrompt: ({ cfg, accountId, payload }) =>
shouldSuppressLocalMatrixExecApprovalPrompt({
cfg,
accountId,
payload,
}),
...createRuntimeOutboundDelegates({
getRuntime: loadMatrixChannelRuntime,
sendText: {
resolve: (runtime) => runtime.matrixOutbound.sendText,
unavailableMessage: "Matrix outbound text delivery is unavailable",
},
sendMedia: {
resolve: (runtime) => runtime.matrixOutbound.sendMedia,
unavailableMessage: "Matrix outbound media delivery is unavailable",
},
sendPoll: {
resolve: (runtime) => runtime.matrixOutbound.sendPoll,
unavailableMessage: "Matrix outbound poll delivery is unavailable",
},
}),
};
const matrixMessageAdapter = createChannelMessageAdapterFromOutbound({
id: "matrix",
outbound: matrixChannelOutbound,
live: {
capabilities: {
draftPreview: true,
previewFinalization: true,
progressUpdates: true,
quietFinalization: true,
},
finalizer: {
capabilities: {
finalEdit: true,
normalFallback: true,
discardPending: true,
previewReceipt: true,
},
},
},
});
export const matrixPlugin: ChannelPlugin<ResolvedMatrixAccount, MatrixProbe> =
createChatChannelPlugin<ResolvedMatrixAccount, MatrixProbe>({
base: {
@@ -416,6 +476,7 @@ export const matrixPlugin: ChannelPlugin<ResolvedMatrixAccount, MatrixProbe> =
}),
resolver: matrixResolverAdapter,
actions: matrixMessageActions,
message: matrixMessageAdapter,
secrets: {
secretTargetRegistryEntries,
collectRuntimeConfigAssignments,
@@ -580,31 +641,5 @@ export const matrixPlugin: ChannelPlugin<ResolvedMatrixAccount, MatrixProbe> =
};
},
},
outbound: {
deliveryMode: "direct",
chunker: chunkTextForOutbound,
chunkerMode: "markdown",
textChunkLimit: 4000,
shouldSuppressLocalPayloadPrompt: ({ cfg, accountId, payload }) =>
shouldSuppressLocalMatrixExecApprovalPrompt({
cfg,
accountId,
payload,
}),
...createRuntimeOutboundDelegates({
getRuntime: loadMatrixChannelRuntime,
sendText: {
resolve: (runtime) => runtime.matrixOutbound.sendText,
unavailableMessage: "Matrix outbound text delivery is unavailable",
},
sendMedia: {
resolve: (runtime) => runtime.matrixOutbound.sendMedia,
unavailableMessage: "Matrix outbound media delivery is unavailable",
},
sendPoll: {
resolve: (runtime) => runtime.matrixOutbound.sendPoll,
unavailableMessage: "Matrix outbound poll delivery is unavailable",
},
}),
},
outbound: matrixChannelOutbound,
});

7
extensions/matrix/src/matrix/draft-stream.test.ts
View File

@@ -64,7 +64,12 @@ const sendModuleMocks = vi.hoisted(() => {
messageId: eventId ?? "unknown",
roomId,
primaryMessageId: eventId ?? "unknown",
messageIds: eventId ? [eventId] : [],
receipt: {
...(eventId ? { primaryPlatformMessageId: eventId } : {}),
platformMessageIds: eventId ? [eventId] : [],
parts: eventId ? [{ platformMessageId: eventId, kind: "text" as const, index: 0 }] : [],
sentAt: 123,
},
};
},
);

158
extensions/matrix/src/matrix/monitor/handler.ts
View File

@@ -1,3 +1,9 @@
import {
createPreviewMessageReceipt,
defineFinalizableLivePreviewAdapter,
deliverWithFinalizableLivePreviewAdapter,
type MessageReceipt,
} from "openclaw/plugin-sdk/channel-message";
import {
createChannelProgressDraftGate,
formatChannelProgressDraftLine,
@@ -894,14 +900,14 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
return undefined;
}
const _messageId = event.event_id ?? "";
const _threadRootId = resolveMatrixThreadRootId({ event, content });
const messageId = event.event_id ?? "";
const threadRootId = resolveMatrixThreadRootId({ event, content });
const thread = resolveMatrixThreadRouting({
isDirectMessage,
threadReplies,
dmThreadReplies,
messageId: _messageId,
threadRootId: _threadRootId,
messageId,
threadRootId,
});
const {
route: _route,
@@ -1001,7 +1007,7 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
sender: senderId,
body: pendingHistoryBody,
timestamp: eventTs ?? undefined,
messageId: _messageId,
messageId,
};
roomHistoryTracker.recordPending(roomId, pendingEntry);
}
@@ -1116,7 +1122,7 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
sender: senderName,
body: bodyText,
timestamp: eventTs ?? undefined,
messageId: _messageId,
messageId,
})
: undefined;
const inboundHistory = preparedTrigger?.history;
@@ -1139,9 +1145,9 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
commandBodyText,
media,
locationPayload,
messageId: _messageId,
messageId,
triggerSnapshot,
threadRootId: _threadRootId,
threadRootId,
thread,
effectiveAllowFrom,
effectiveGroupAllowFrom,
@@ -1194,9 +1200,9 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
commandBodyText,
media,
locationPayload,
messageId: _messageId,
messageId,
triggerSnapshot,
threadRootId: _threadRootId,
threadRootId,
thread,
effectiveGroupAllowFrom,
effectiveRoomUsers,
@@ -1233,8 +1239,8 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
kind,
senderAllowed: isRoomContextSenderAllowed(contextSenderId),
}).include;
let threadContext = _threadRootId
? await resolveThreadContext({ roomId, threadRootId: _threadRootId })
let threadContext = threadRootId
? await resolveThreadContext({ roomId, threadRootId })
: undefined;
let threadContextBlockedByPolicy = false;
if (
@@ -1246,7 +1252,7 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
threadContext = undefined;
}
let replyContext: Awaited<ReturnType<typeof resolveReplyContext>> | undefined;
if (replyToEventId && replyToEventId === _threadRootId && threadContext?.summary) {
if (replyToEventId && replyToEventId === threadRootId && threadContext?.summary) {
replyContext = {
replyToBody: threadContext.summary,
replyToSender: threadContext.senderLabel,
@@ -1254,7 +1260,7 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
};
} else if (
replyToEventId &&
replyToEventId === _threadRootId &&
replyToEventId === threadRootId &&
threadContextBlockedByPolicy
) {
replyContext = await resolveReplyContext({ roomId, eventId: replyToEventId });
@@ -1273,7 +1279,7 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
const roomInfo = isRoom ? await getRoomInfo(roomId) : undefined;
const roomName = roomInfo?.name;
const envelopeFrom = isDirectMessage ? senderName : (roomName ?? roomId);
const textWithId = `${bodyText}\n[matrix event id: ${_messageId} room: ${roomId}]`;
const textWithId = `${bodyText}\n[matrix event id: ${messageId} room: ${roomId}]`;
const storePath = core.channel.session.resolveStorePath(cfg.session?.store, {
agentId: _route.agentId,
});
@@ -1330,7 +1336,7 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
Provider: "matrix" as const,
Surface: "matrix" as const,
WasMentioned: isRoom ? wasMentioned : undefined,
MessageSid: _messageId,
MessageSid: messageId,
ReplyToId: threadTarget ? undefined : (replyToEventId ?? undefined),
ReplyToBody: replyContext?.replyToBody,
ReplyToSender: replyContext?.replyToSender,
@@ -1377,22 +1383,22 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
shouldBypassMention,
}),
);
if (shouldAckReaction() && _messageId) {
if (shouldAckReaction() && messageId) {
loadMatrixSendModule()
.then(({ reactMatrixMessage }) =>
reactMatrixMessage(roomId, _messageId, ackReaction, client),
reactMatrixMessage(roomId, messageId, ackReaction, client),
)
.catch((err) => {
logVerboseMessage(`matrix react failed for room ${roomId}: ${String(err)}`);
});
}
if (_messageId) {
if (messageId) {
loadMatrixSendModule()
.then(({ sendReadReceiptMatrix }) => sendReadReceiptMatrix(roomId, _messageId, client))
.then(({ sendReadReceiptMatrix }) => sendReadReceiptMatrix(roomId, messageId, client))
.catch((err) => {
logVerboseMessage(
`matrix: read receipt failed room=${roomId} id=${_messageId}: ${String(err)}`,
`matrix: read receipt failed room=${roomId} id=${messageId}: ${String(err)}`,
);
});
}
@@ -1443,7 +1449,7 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
const draftStreamingEnabled = streaming !== "off";
const quietDraftStreaming = streaming === "quiet" || streaming === "progress";
const progressDraftStreaming = streaming === "progress";
const draftReplyToId = replyToMode !== "off" && !threadTarget ? _messageId : undefined;
const draftReplyToId = replyToMode !== "off" && !threadTarget ? messageId : undefined;
const draftStream: MatrixDraftStreamHandle | undefined = draftStreamingEnabled
? await loadMatrixDraftStream().then(({ createMatrixDraftStream }) =>
createMatrixDraftStream({
@@ -1785,39 +1791,77 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
!payloadReplyMismatch &&
!mustDeliverFinalNormally
) {
try {
const requiresFinalEdit =
quietDraftStreaming || !draftStream.matchesPreparedText(payload.text);
if (requiresFinalEdit) {
const { editMessageMatrix } = await loadMatrixSendModule();
await editMessageMatrix(roomId, draftEventId, payload.text, {
client,
const finalPreviewText = payload.text;
await deliverWithFinalizableLivePreviewAdapter<
ReplyPayload,
string,
{
text: string;
finalizeLive: boolean;
extraContent?: Record<string, unknown>;
}
>({
kind: "final",
payload,
adapter: defineFinalizableLivePreviewAdapter({
draft: {
flush: async () => {},
clear: async () => {},
discardPending: async () => {},
id: () => draftEventId,
},
buildFinalEdit: () => ({
text: finalPreviewText,
finalizeLive: !(
quietDraftStreaming || !draftStream.matchesPreparedText(finalPreviewText)
),
...(quietDraftStreaming
? { extraContent: buildMatrixFinalizedPreviewContent() }
: {}),
}),
editFinal: async (_draftEventId, edit) => {
if (edit.finalizeLive) {
if (!(await draftStream.finalizeLive())) {
throw new Error("Matrix draft live finalize failed");
}
return;
}
const { editMessageMatrix } = await loadMatrixSendModule();
await editMessageMatrix(roomId, _draftEventId, edit.text, {
client,
cfg,
threadId: threadTarget,
accountId: _route.accountId,
extraContent: edit.extraContent,
});
},
createPreviewReceipt: (id): MessageReceipt =>
createPreviewMessageReceipt({
id,
...(threadTarget ? { threadId: threadTarget } : {}),
...(currentDraftReplyToId ? { replyToId: currentDraftReplyToId } : {}),
}),
logPreviewEditFailure: (err) => {
logVerboseMessage(`matrix: preview final edit failed: ${String(err)}`);
},
}),
deliverNormally: async () => {
await redactMatrixDraftEvent(client, roomId, draftEventId);
await deliverMatrixReplies({
cfg,
replies: [payload],
roomId,
client,
runtime,
textLimit,
replyToMode,
threadId: threadTarget,
accountId: _route.accountId,
extraContent: quietDraftStreaming
? buildMatrixFinalizedPreviewContent()
: undefined,
mediaLocalRoots,
tableMode,
});
} else if (!(await draftStream.finalizeLive())) {
throw new Error("Matrix draft live finalize failed");
}
} catch {
await redactMatrixDraftEvent(client, roomId, draftEventId);
await deliverMatrixReplies({
cfg,
replies: [payload],
roomId,
client,
runtime,
textLimit,
replyToMode,
threadId: threadTarget,
accountId: _route.accountId,
mediaLocalRoots,
tableMode,
});
}
},
});
draftConsumed = true;
} else if (draftEventId && hasMedia && !payloadReplyMismatch) {
let textEditOk = !mustDeliverFinalNormally;
@@ -1968,7 +2012,7 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
raw: event,
adapter: {
ingest: () => ({
id: _messageId,
id: messageId,
rawText: bodyText,
textForAgent: ctxPayload.BodyForAgent,
textForCommands: ctxPayload.CommandBody,
@@ -2108,13 +2152,13 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
if (finalReplyDeliveryFailed) {
if (retryableReplyDeliveryFailed) {
logVerboseMessage(
`matrix: final reply delivery failed room=${roomId} id=${_messageId}; leaving event uncommitted`,
`matrix: final reply delivery failed room=${roomId} id=${messageId}; leaving event uncommitted`,
);
// Explicit retryable failures reopen replay so the same history can be retried.
return;
}
logVerboseMessage(
`matrix: final reply delivery failed room=${roomId} id=${_messageId}; keeping replay committed`,
`matrix: final reply delivery failed room=${roomId} id=${messageId}; keeping replay committed`,
);
await commitInboundEventIfClaimed();
return;
@@ -2122,13 +2166,13 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
if (!queuedFinal && nonFinalReplyDeliveryFailed) {
if (retryableReplyDeliveryFailed) {
logVerboseMessage(
`matrix: non-final reply delivery failed room=${roomId} id=${_messageId}; leaving event uncommitted`,
`matrix: non-final reply delivery failed room=${roomId} id=${messageId}; leaving event uncommitted`,
);
// Explicit retryable failures reopen replay.
return;
}
logVerboseMessage(
`matrix: non-final reply delivery failed room=${roomId} id=${_messageId}; keeping replay committed`,
`matrix: non-final reply delivery failed room=${roomId} id=${messageId}; keeping replay committed`,
);
await commitInboundEventIfClaimed();
return;
@@ -2137,7 +2181,7 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
// Only advance to the snapshot position — messages added during async processing remain
// visible for the next trigger.
if (isRoom && triggerSnapshot) {
roomHistoryTracker.consumeHistory(_route.agentId, roomId, triggerSnapshot, _messageId);
roomHistoryTracker.consumeHistory(_route.agentId, roomId, triggerSnapshot, messageId);
}
if (!hasFinalInboundReplyDispatch({ queuedFinal, counts })) {
await commitInboundEventIfClaimed();

17
extensions/matrix/src/matrix/send.test.ts
View File

@@ -627,7 +627,15 @@ describe("sendMessageMatrix threads", () => {
roomId: "!room:example",
primaryMessageId: "$m1",
messageId: "$m3",
messageIds: ["$m1", "$m2", "$m3"],
receipt: {
primaryPlatformMessageId: "$m1",
platformMessageIds: ["$m1", "$m2", "$m3"],
parts: [
expect.objectContaining({ platformMessageId: "$m1", kind: "text" }),
expect.objectContaining({ platformMessageId: "$m2", kind: "text" }),
expect.objectContaining({ platformMessageId: "$m3", kind: "text" }),
],
},
});
});
@@ -720,7 +728,7 @@ describe("sendSingleTextMessageMatrix", () => {
it("merges extra content fields into single-event sends", async () => {
const { client, sendMessage } = makeClient();
await sendSingleTextMessageMatrix("room:!room:example", "done", {
const result = await sendSingleTextMessageMatrix("room:!room:example", "done", {
client,
cfg: {} as never,
extraContent: { [MATRIX_OPENCLAW_FINALIZED_PREVIEW_KEY]: true },
@@ -730,6 +738,11 @@ describe("sendSingleTextMessageMatrix", () => {
body: "done",
[MATRIX_OPENCLAW_FINALIZED_PREVIEW_KEY]: true,
});
expect(result.receipt).toMatchObject({
primaryPlatformMessageId: "evt1",
platformMessageIds: ["evt1"],
parts: [expect.objectContaining({ platformMessageId: "evt1", kind: "text" })],
});
});
});

52
extensions/matrix/src/matrix/send.ts
View File

@@ -1,3 +1,7 @@
import {
createMessageReceiptFromOutboundResults,
type MessageReceiptPartKind,
} from "openclaw/plugin-sdk/channel-message";
import type { MarkdownTableMode } from "openclaw/plugin-sdk/markdown-table-runtime";
import { requireRuntimeConfig } from "openclaw/plugin-sdk/plugin-config-runtime";
import type { PollInput } from "../runtime-api.js";
@@ -66,6 +70,25 @@ type MatrixClientResolveOpts = {
accountId?: string | null;
};
function createMatrixSendReceipt(params: {
roomId: string;
platformMessageIds: readonly string[];
kind: MessageReceiptPartKind;
replyToId?: string;
threadId?: string | null;
}) {
return createMessageReceiptFromOutboundResults({
kind: params.kind,
...(params.replyToId ? { replyToId: params.replyToId } : {}),
...(params.threadId ? { threadId: params.threadId } : {}),
results: params.platformMessageIds.map((messageId) => ({
channel: "matrix",
messageId,
roomId: params.roomId,
})),
});
}
function isMatrixClient(value: MatrixClient | MatrixClientResolveOpts): value is MatrixClient {
return typeof (value as { sendEvent?: unknown }).sendEvent === "function";
}
@@ -219,8 +242,9 @@ export async function sendMessageMatrix(
return eventId;
};
const messageIds: string[] = [];
const platformMessageIds: string[] = [];
let lastMessageId = "";
let receiptKind: MessageReceiptPartKind = "text";
if (opts.mediaUrl) {
const maxBytes = resolveMediaMaxBytes(opts.accountId, cfg);
const media = await loadOutboundMediaFromUrl(opts.mediaUrl, {
@@ -246,6 +270,7 @@ export async function sendMessageMatrix(
fileName: media.fileName,
});
const msgtype = useVoice ? MsgType.Audio : baseMsgType;
receiptKind = useVoice ? "voice" : "media";
const isImage = msgtype === MsgType.Image;
const imageInfo = isImage
? await prepareImageInfo({
@@ -278,7 +303,7 @@ export async function sendMessageMatrix(
const eventId = await sendContent(content);
lastMessageId = eventId ?? lastMessageId;
if (eventId) {
messageIds.push(eventId);
platformMessageIds.push(eventId);
}
const textChunks = useVoice ? chunks : rest;
// Voice messages use a generic media body ("Voice message"), so keep any
@@ -298,7 +323,7 @@ export async function sendMessageMatrix(
const followupEventId = await sendContent(followup);
lastMessageId = followupEventId ?? lastMessageId;
if (followupEventId) {
messageIds.push(followupEventId);
platformMessageIds.push(followupEventId);
}
}
} else {
@@ -316,7 +341,7 @@ export async function sendMessageMatrix(
const eventId = await sendContent(content);
lastMessageId = eventId ?? lastMessageId;
if (eventId) {
messageIds.push(eventId);
platformMessageIds.push(eventId);
}
}
}
@@ -324,8 +349,14 @@ export async function sendMessageMatrix(
return {
messageId: lastMessageId || "unknown",
roomId,
primaryMessageId: messageIds[0] ?? (lastMessageId || "unknown"),
messageIds,
primaryMessageId: platformMessageIds[0] ?? (lastMessageId || "unknown"),
receipt: createMatrixSendReceipt({
roomId,
platformMessageIds,
kind: receiptKind,
replyToId: opts.replyToId,
threadId,
}),
};
},
);
@@ -474,11 +505,18 @@ export async function sendSingleTextMessageMatrix(
(content as Record<string, unknown>)[MSC4357_LIVE_KEY] = {};
}
const eventId = await client.sendMessage(resolvedRoom, content);
const platformMessageIds = eventId ? [eventId] : [];
return {
messageId: eventId ?? "unknown",
roomId: resolvedRoom,
primaryMessageId: eventId ?? "unknown",
messageIds: eventId ? [eventId] : [],
receipt: createMatrixSendReceipt({
roomId: resolvedRoom,
platformMessageIds,
kind: "text",
replyToId: opts.replyToId,
threadId: normalizedThreadId,
}),
};
},
);

3
extensions/matrix/src/matrix/send/types.ts
View File

@@ -1,3 +1,4 @@
import type { MessageReceipt } from "openclaw/plugin-sdk/channel-message";
import type { CoreConfig } from "../../types.js";
import { MATRIX_ANNOTATION_RELATION_TYPE, MATRIX_REACTION_EVENT_TYPE } from "../reaction-common.js";
import type {
@@ -79,7 +80,7 @@ export type MatrixSendResult = {
messageId: string;
roomId: string;
primaryMessageId?: string;
messageIds?: string[];
receipt: MessageReceipt;
};
export type MatrixSendOpts = {

2
extensions/matrix/src/runtime-api.ts
View File

@@ -90,7 +90,7 @@ export {
export { resolveOutboundSendDep } from "openclaw/plugin-sdk/outbound-send-deps";
export { resolveAgentIdFromSessionKey } from "openclaw/plugin-sdk/routing";
export { chunkTextForOutbound } from "openclaw/plugin-sdk/text-chunking";
export { createChannelReplyPipeline } from "openclaw/plugin-sdk/channel-reply-pipeline";
export { createChannelMessageReplyPipeline } from "openclaw/plugin-sdk/channel-message";
export { loadOutboundMediaFromUrl } from "openclaw/plugin-sdk/outbound-media";
export { normalizePollInput, type PollInput } from "openclaw/plugin-sdk/poll-runtime";
export { writeJsonFileAtomically } from "openclaw/plugin-sdk/json-store";