1-import { describe, expect, it, vi, beforeAll, beforeEach } from "vitest";
1+import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
2+import type { ChannelOutboundAdapter } from "../../channels/plugins/types.js";
23import type { OpenClawConfig } from "../../config/config.js";
4+import { createEmptyPluginRegistry } from "../../plugins/registry.js";
5+import {
6+releasePinnedPluginChannelRegistry,
7+setActivePluginRegistry,
8+} from "../../plugins/runtime.js";
9+import { createOutboundTestPlugin, createTestRegistry } from "../../test-utils/channel-plugins.js";
310import { drainPendingDeliveries, type DeliverFn, loadPendingDeliveries } from "./delivery-queue.js";
411import {
512createRecoveryLog,
@@ -8,6 +15,40 @@ import {
815916let deliverOutboundPayloads: typeof import("./deliver.js").deliverOutboundPayloads;
101718+type MatrixSendFn = (
19+to: string,
20+text: string,
21+options?: Record<string, unknown>,
22+) => Promise<{ messageId: string } & Record<string, unknown>>;
23+24+function resolveMatrixSender(
25+deps: Parameters<NonNullable<ChannelOutboundAdapter["sendText"]>>[0]["deps"],
26+): MatrixSendFn {
27+const sender = deps?.matrix;
28+if (typeof sender !== "function") {
29+throw new Error("missing matrix sender");
30+}
31+return sender as MatrixSendFn;
32+}
33+34+function withMatrixChannel(result: Awaited<ReturnType<MatrixSendFn>>) {
35+return {
36+channel: "matrix" as const,
37+ ...result,
38+};
39+}
40+41+const matrixOutboundForQueueTest: ChannelOutboundAdapter = {
42+deliveryMode: "direct",
43+sendText: async ({ cfg, to, text, accountId, deps }) =>
44+withMatrixChannel(
45+await resolveMatrixSender(deps)(to, text, {
46+ cfg,
47+accountId: accountId ?? undefined,
48+}),
49+),
50+};
51+1152async function drainMatrixReconnect(opts: { deliver: DeliverFn; stateDir: string }): Promise<void> {
1253await drainPendingDeliveries({
1354drainKey: "matrix:reconnect-test",
@@ -51,6 +92,20 @@ describe("deliverOutboundPayloads queue integration: mid-batch failure with send
51925293beforeEach(() => {
5394tmpDir = fixtures.tmpDir();
95+setActivePluginRegistry(
96+createTestRegistry([
97+{
98+pluginId: "matrix",
99+source: "test",
100+plugin: createOutboundTestPlugin({ id: "matrix", outbound: matrixOutboundForQueueTest }),
101+},
102+]),
103+);
104+});
105+106+afterEach(() => {
107+releasePinnedPluginChannelRegistry();
108+setActivePluginRegistry(createEmptyPluginRegistry());
54109});
5511056111it("advances queued entry to unknown_after_send when a later payload fails after an earlier one succeeded", async () => {
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。