diff --git a/extensions/feishu/src/channel.ts b/extensions/feishu/src/channel.ts index 6bd5f2028432..9b174b762f56 100644 --- a/extensions/feishu/src/channel.ts +++ b/extensions/feishu/src/channel.ts @@ -29,6 +29,7 @@ import { createChannelDirectoryAdapter, createRuntimeDirectoryLiveAdapter, } from "openclaw/plugin-sdk/directory-runtime"; +import { PlatformMessageNotDispatchedError } from "openclaw/plugin-sdk/error-runtime"; import { legacyInteractiveReplyToPresentation, normalizeLegacyInteractiveReply, @@ -180,6 +181,38 @@ const loadFeishuChannelRuntime = createLazyRuntimeNamedExport( "feishuChannelRuntime", ); +async function resolveFeishuMessageSender(params: { + resolve: ( + runtime: Awaited>, + ) => TSender | null | undefined; + unavailableMessage: string; +}): Promise { + try { + const sender = params.resolve(await loadFeishuChannelRuntime()); + if (sender) { + return sender; + } + throw new Error(params.unavailableMessage); + } catch (error) { + if (error instanceof PlatformMessageNotDispatchedError) { + throw error; + } + throw new PlatformMessageNotDispatchedError(params.unavailableMessage, { cause: error }); + } +} + +const resolveFeishuTextSender = () => + resolveFeishuMessageSender({ + resolve: (runtime) => runtime.feishuOutbound.sendText, + unavailableMessage: "Feishu text sending is not available.", + }); + +const resolveFeishuMediaSender = () => + resolveFeishuMessageSender({ + resolve: (runtime) => runtime.feishuOutbound.sendMedia, + unavailableMessage: "Feishu media sending is not available.", + }); + function toFeishuMessageSendResult( result: { messageId?: string; chatId?: string; receipt?: ChannelMessageSendResult["receipt"] }, kind: MessageReceiptPartKind, @@ -206,12 +239,19 @@ const feishuMessageAdapter = defineChannelMessageAdapter({ }, }, send: { + lifecycle: { + // Resolve process-stable runtime methods before core records platform-send start. + // Provider invocation stays below so a lost provider result remains ambiguous. + beforeSendAttempt: async (ctx) => { + if (ctx.kind === "text") { + await resolveFeishuTextSender(); + } else if (ctx.kind === "media") { + await resolveFeishuMediaSender(); + } + }, + }, text: async (ctx) => { - const runtime = await loadFeishuChannelRuntime(); - const sendText = runtime.feishuOutbound.sendText; - if (!sendText) { - throw new Error("Feishu text sending is not available."); - } + const sendText = await resolveFeishuTextSender(); const { onDeliveryResult, ...outboundCtx } = ctx; const result = await sendText({ ...outboundCtx, @@ -226,11 +266,7 @@ const feishuMessageAdapter = defineChannelMessageAdapter({ return toFeishuMessageSendResult(result, "text"); }, media: async (ctx) => { - const runtime = await loadFeishuChannelRuntime(); - const sendMedia = runtime.feishuOutbound.sendMedia; - if (!sendMedia) { - throw new Error("Feishu media sending is not available."); - } + const sendMedia = await resolveFeishuMediaSender(); const { onDeliveryResult, ...outboundCtx } = ctx; const result = await sendMedia({ ...outboundCtx, diff --git a/extensions/feishu/src/outbound-delivery.test.ts b/extensions/feishu/src/outbound-delivery.test.ts index 8d19027b94df..2868644112ec 100644 --- a/extensions/feishu/src/outbound-delivery.test.ts +++ b/extensions/feishu/src/outbound-delivery.test.ts @@ -1,4 +1,6 @@ // Feishu tests cover the shared outbound delivery path. +import path from "node:path"; +import { DatabaseSync } from "node:sqlite"; import { sendDurableMessageBatch } from "openclaw/plugin-sdk/channel-outbound"; import { createOutboundTestPlugin, @@ -7,6 +9,8 @@ import { resetGlobalHookRunner, setActivePluginRegistry, } from "openclaw/plugin-sdk/channel-test-helpers"; +import { drainPendingDeliveries } from "openclaw/plugin-sdk/delivery-queue-runtime"; +import { withStateDirEnv } from "openclaw/plugin-sdk/test-env"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; const sendMediaFeishuMock = vi.hoisted(() => vi.fn()); @@ -19,14 +23,47 @@ vi.mock("./media.js", () => ({ })); vi.mock("./send.js", () => ({ + editMessageFeishu: vi.fn(), + getMessageFeishu: vi.fn(), sendCardFeishu: sendCardFeishuMock, sendMarkdownCardFeishu: vi.fn(), sendMessageFeishu: sendMessageFeishuMock, sendStructuredCardFeishu: vi.fn(), })); +import { feishuPlugin } from "./channel.js"; +import { feishuChannelRuntime } from "./channel.runtime.js"; import { feishuOutbound } from "./outbound.js"; +type DeliveryQueueRow = { + status: string; + recovery_state: string | null; + platform_send_started_at: number | null; +}; + +const completionRetention = { + idPrefix: "feishu-direct-", + maxAgeMs: 60_000, + maxEntries: 10, +} as const; + +function readDeliveryQueueRow(stateDir: string, id: string): DeliveryQueueRow | undefined { + const database = new DatabaseSync(path.join(stateDir, "state", "openclaw.sqlite"), { + readOnly: true, + }); + try { + return database + .prepare( + `SELECT status, recovery_state, platform_send_started_at + FROM delivery_queue_entries + WHERE queue_name = 'outbound-prepared-v1' AND id = ?`, + ) + .get(id) as DeliveryQueueRow | undefined; + } finally { + database.close(); + } +} + describe("Feishu outbound shared delivery", () => { beforeEach(() => { let textMessageIndex = 0; @@ -119,4 +156,110 @@ describe("Feishu outbound shared delivery", () => { expect(deliveredText).toContain("account-0-"); expect(deliveredText).toContain("account-399-"); }); + + it("replays a queued direct message after Feishu runtime availability is restored", async () => { + const originalSendText = feishuChannelRuntime.feishuOutbound.sendText; + if (!originalSendText) { + throw new Error("Expected Feishu runtime sendText"); + } + const deliveryIntentId = "feishu-direct-runtime-availability"; + + setActivePluginRegistry( + createTestRegistry([{ pluginId: "feishu", plugin: feishuPlugin, source: "test" }]), + ); + feishuChannelRuntime.feishuOutbound.sendText = undefined; + + try { + await withStateDirEnv("openclaw-feishu-runtime-availability-", async ({ stateDir }) => { + const initial = await sendDurableMessageBatch({ + cfg: {}, + channel: "feishu", + to: "chat_1", + accountId: "default", + durability: "required", + deliveryIntentId, + completionRetention, + maxRetries: 2, + payloads: [{ text: "retry after runtime restoration" }], + }); + + expect(initial.status).toBe("failed"); + expect(sendMessageFeishuMock).not.toHaveBeenCalled(); + expect(readDeliveryQueueRow(stateDir, deliveryIntentId)).toMatchObject({ + status: "pending", + recovery_state: null, + platform_send_started_at: null, + }); + + feishuChannelRuntime.feishuOutbound.sendText = originalSendText; + await drainPendingDeliveries({ + drainKey: "feishu:default", + logLabel: "Feishu runtime availability recovery", + cfg: {}, + stateDir, + log: { info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + selectEntry: (entry) => ({ + match: entry.channel === "feishu", + bypassBackoff: true, + }), + }); + + expect(sendMessageFeishuMock).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ + to: "chat_1", + text: "retry after runtime restoration", + }), + ); + expect(readDeliveryQueueRow(stateDir, deliveryIntentId)?.status).toBe("completed"); + }); + } finally { + feishuChannelRuntime.feishuOutbound.sendText = originalSendText; + } + }); + + it("does not replay a Feishu provider call after dispatch may have begun", async () => { + const deliveryIntentId = "feishu-direct-ambiguous-provider-result"; + sendMessageFeishuMock.mockRejectedValueOnce(new Error("Feishu provider result was lost")); + setActivePluginRegistry( + createTestRegistry([{ pluginId: "feishu", plugin: feishuPlugin, source: "test" }]), + ); + + await withStateDirEnv("openclaw-feishu-ambiguous-provider-", async ({ stateDir }) => { + const initial = await sendDurableMessageBatch({ + cfg: {}, + channel: "feishu", + to: "chat_1", + accountId: "default", + durability: "required", + deliveryIntentId, + completionRetention, + maxRetries: 2, + payloads: [{ text: "do not replay an ambiguous provider call" }], + }); + + expect(initial.status).toBe("failed"); + expect(sendMessageFeishuMock).toHaveBeenCalledOnce(); + expect(readDeliveryQueueRow(stateDir, deliveryIntentId)).toMatchObject({ + status: "pending", + recovery_state: "send_attempt_started", + }); + expect(readDeliveryQueueRow(stateDir, deliveryIntentId)?.platform_send_started_at).toEqual( + expect.any(Number), + ); + + await drainPendingDeliveries({ + drainKey: "feishu:default", + logLabel: "Feishu ambiguous provider recovery", + cfg: {}, + stateDir, + log: { info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + selectEntry: (entry) => ({ + match: entry.channel === "feishu", + bypassBackoff: true, + }), + }); + + expect(sendMessageFeishuMock).toHaveBeenCalledOnce(); + }); + }); }); diff --git a/src/channels/plugins/contracts/plugins-core.loader.contract.test.ts b/src/channels/plugins/contracts/plugins-core.loader.contract.test.ts index 8a94ce1de47d..fcaa45c0eeec 100644 --- a/src/channels/plugins/contracts/plugins-core.loader.contract.test.ts +++ b/src/channels/plugins/contracts/plugins-core.loader.contract.test.ts @@ -1,6 +1,7 @@ // Plugins core loader contract tests cover channel plugin loader setup and teardown behavior. import { afterEach, beforeEach, describe, expect, it } from "vitest"; import { setActivePluginRegistry } from "../../../plugins/runtime.js"; +import { withPluginRuntimeRegistryScope } from "../../../plugins/runtime/gateway-request-scope.js"; import { createChannelTestPluginBase, createOutboundTestPlugin, @@ -103,6 +104,16 @@ describe("channel plugin loader", () => { setActivePluginRegistry(emptyRegistry); }); + it("prefers the registry scoped to a bootstrapped channel handler", async () => { + setActivePluginRegistry(emptyRegistry); + + const loaded = await withPluginRuntimeRegistryScope(registryWithDemoLoader, () => + loadChannelOutboundAdapter("demo-loader"), + ); + + expect(loaded).toBe(demoOutbound); + }); + it.each([ { name: "loads channel plugins from the active registry", diff --git a/src/channels/plugins/registry-loader.ts b/src/channels/plugins/registry-loader.ts index 9d40b4a644e3..3bb60f21b78b 100644 --- a/src/channels/plugins/registry-loader.ts +++ b/src/channels/plugins/registry-loader.ts @@ -5,6 +5,7 @@ */ import type { PluginChannelRegistration } from "../../plugins/registry-types.js"; import { getActivePluginRegistry } from "../../plugins/runtime.js"; +import { getPluginRuntimeGatewayRequestScope } from "../../plugins/runtime/gateway-request-scope.js"; import type { ChannelId } from "./channel-id.types.js"; type ChannelRegistryValueResolver = ( @@ -25,6 +26,8 @@ export function createChannelRegistryLoader( return pluginEntry ? resolveValue(pluginEntry) : undefined; }; - return resolveFromRegistry(getActivePluginRegistry()); + return resolveFromRegistry( + getPluginRuntimeGatewayRequestScope()?.pluginRegistry ?? getActivePluginRegistry(), + ); }; } diff --git a/src/channels/plugins/runtime-forwarders.test.ts b/src/channels/plugins/runtime-forwarders.test.ts index f88effe1d4d7..8a74439a00f3 100644 --- a/src/channels/plugins/runtime-forwarders.test.ts +++ b/src/channels/plugins/runtime-forwarders.test.ts @@ -1,5 +1,6 @@ // Runtime forwarder tests cover channel plugin runtime method delegation and fallback handling. import { describe, expect, it, vi } from "vitest"; +import { PlatformMessageNotDispatchedError } from "../../infra/outbound/deliver-types.js"; import { createRuntimeDirectoryLiveAdapter, createRuntimeOutboundDelegates, @@ -68,7 +69,7 @@ describe("createRuntimeOutboundDelegates", () => { expect(sendText).toHaveBeenCalled(); }); - it("throws the configured unavailable message", async () => { + it("classifies unavailable outbound runtime methods as definitely not dispatched", async () => { const outbound = createRuntimeOutboundDelegates({ getRuntime: async () => ({ outbound: {} }), sendPoll: { @@ -77,12 +78,34 @@ describe("createRuntimeOutboundDelegates", () => { }, }); - await expect( - outbound.sendPoll?.({ + const error = await outbound + .sendPoll?.({ cfg: {} as never, to: "a", poll: { question: "q", options: ["a"] }, - }), - ).rejects.toThrow("poll unavailable"); + }) + .catch((caught: unknown) => caught); + expect(error).toBeInstanceOf(PlatformMessageNotDispatchedError); + expect(error).toMatchObject({ message: "poll unavailable" }); + }); + + it("classifies outbound runtime loading failures before method dispatch", async () => { + const loadError = new Error("runtime import failed"); + const outbound = createRuntimeOutboundDelegates({ + getRuntime: async () => { + throw loadError; + }, + sendText: { + resolve: (runtime: { sendText?: ChannelOutboundAdapter["sendText"] }) => runtime.sendText, + }, + }); + + await expect( + outbound.sendText?.({ cfg: {} as never, to: "a", text: "hi" }), + ).rejects.toMatchObject({ + name: "PlatformMessageNotDispatchedError", + message: "runtime import failed", + cause: loadError, + }); }); }); diff --git a/src/channels/plugins/runtime-forwarders.ts b/src/channels/plugins/runtime-forwarders.ts index 223763d0eb3b..5eeac325d226 100644 --- a/src/channels/plugins/runtime-forwarders.ts +++ b/src/channels/plugins/runtime-forwarders.ts @@ -3,6 +3,7 @@ * * Creates directory and outbound adapters whose methods delegate to lazily resolved runtimes. */ +import { PlatformMessageNotDispatchedError } from "../../infra/outbound/deliver-types.js"; import type { ChannelDirectoryAdapter, ChannelOutboundAdapter } from "./types.adapters.js"; type MaybePromise = T | Promise; @@ -26,16 +27,29 @@ type SendPayloadParams = Parameters(params: { getRuntime: () => MaybePromise; resolve: (runtime: Runtime) => Fn | null | undefined; + notDispatched?: boolean; unavailableMessage?: string; }): Promise { - const runtime = await params.getRuntime(); - const method = params.resolve(runtime); - if (method) { - return method; + try { + const runtime = await params.getRuntime(); + const method = params.resolve(runtime); + if (method) { + return method; + } + // Fail at call time instead of registration time so optional runtime methods + // can stay absent until the caller actually invokes that capability. + throw new Error(params.unavailableMessage ?? "Runtime method is unavailable"); + } catch (error) { + if (!params.notDispatched || error instanceof PlatformMessageNotDispatchedError) { + throw error; + } + const message = + params.unavailableMessage ?? + (error instanceof Error && error.message.trim() + ? error.message + : "Runtime method is unavailable"); + throw new PlatformMessageNotDispatchedError(message, { cause: error }); } - // Fail at call time instead of registration time so optional runtime methods - // can stay absent until the caller actually invokes that capability. - throw new Error(params.unavailableMessage ?? "Runtime method is unavailable"); } /** @@ -134,6 +148,7 @@ export function createRuntimeOutboundDelegates(params: { await ( await resolveForwardedMethod({ getRuntime: params.getRuntime, + notDispatched: true, resolve: params.sendPayload!.resolve, unavailableMessage: params.sendPayload!.unavailableMessage, }) @@ -144,6 +159,7 @@ export function createRuntimeOutboundDelegates(params: { await ( await resolveForwardedMethod({ getRuntime: params.getRuntime, + notDispatched: true, resolve: params.sendText!.resolve, unavailableMessage: params.sendText!.unavailableMessage, }) @@ -154,6 +170,7 @@ export function createRuntimeOutboundDelegates(params: { await ( await resolveForwardedMethod({ getRuntime: params.getRuntime, + notDispatched: true, resolve: params.sendMedia!.resolve, unavailableMessage: params.sendMedia!.unavailableMessage, }) @@ -164,6 +181,7 @@ export function createRuntimeOutboundDelegates(params: { await ( await resolveForwardedMethod({ getRuntime: params.getRuntime, + notDispatched: true, resolve: params.sendPoll!.resolve, unavailableMessage: params.sendPoll!.unavailableMessage, }) diff --git a/src/cli/send-runtime/channel-outbound-send.test.ts b/src/cli/send-runtime/channel-outbound-send.test.ts index b73f8dacc345..3f8b53f1837c 100644 --- a/src/cli/send-runtime/channel-outbound-send.test.ts +++ b/src/cli/send-runtime/channel-outbound-send.test.ts @@ -1,5 +1,6 @@ // Channel outbound send tests cover CLI send runtime handoff to channel outbound adapters. import { beforeEach, describe, expect, it, vi } from "vitest"; +import { PlatformMessageNotDispatchedError } from "../../infra/outbound/deliver-types.js"; const mocks = vi.hoisted(() => ({ loadChannelOutboundAdapter: vi.fn(), @@ -23,6 +24,32 @@ describe("createChannelOutboundRuntimeSend", () => { return params; } + it.each(["discord", "telegram"] as const)( + "classifies unavailable %s adapters as definitely not dispatched", + async (channelId) => { + mocks.loadChannelOutboundAdapter.mockResolvedValue(undefined); + const unavailableMessage = `${channelId} outbound adapter is unavailable.`; + + const { createChannelOutboundRuntimeSend } = await import("./channel-outbound-send.js"); + const runtimeSend = createChannelOutboundRuntimeSend({ + channelId, + unavailableMessage, + }); + + const error = await runtimeSend + .sendMessage("target", "hello", { cfg: {} }) + .catch((caught: unknown) => caught); + expect(error).toBeInstanceOf(PlatformMessageNotDispatchedError); + expect(error).toMatchObject({ + name: "PlatformMessageNotDispatchedError", + message: unavailableMessage, + cause: expect.objectContaining({ + message: unavailableMessage, + }), + }); + }, + ); + it("routes media sends through sendMedia and preserves media access", async () => { const sendMedia = vi.fn(async () => ({ channel: "whatsapp", messageId: "wa-1" })); mocks.loadChannelOutboundAdapter.mockResolvedValue({ diff --git a/src/cli/send-runtime/channel-outbound-send.ts b/src/cli/send-runtime/channel-outbound-send.ts index 96c9400ccff0..4109f87944ac 100644 --- a/src/cli/send-runtime/channel-outbound-send.ts +++ b/src/cli/send-runtime/channel-outbound-send.ts @@ -4,6 +4,7 @@ import { loadChannelOutboundAdapter } from "../../channels/plugins/outbound/load import type { ChannelId } from "../../channels/plugins/types.public.js"; import { getRuntimeConfig } from "../../config/config.js"; import type { OpenClawConfig } from "../../config/types.openclaw.js"; +import { PlatformMessageNotDispatchedError } from "../../infra/outbound/deliver-types.js"; import type { OutboundDeliveryFormattingOptions } from "../../infra/outbound/formatting.js"; import type { OutboundMediaAccess } from "../../media/load-options.js"; @@ -96,7 +97,8 @@ export function createChannelOutboundRuntimeSend(params: { return await outbound.sendMedia(buildContext()); } if (!outbound?.sendText) { - throw new Error(params.unavailableMessage); + const cause = new Error(params.unavailableMessage); + throw new PlatformMessageNotDispatchedError(params.unavailableMessage, { cause }); } return await outbound.sendText(buildContext()); }, diff --git a/src/gateway/test-helpers.plugin-registry.ts b/src/gateway/test-helpers.plugin-registry.ts index a4e7651b2ca7..52d37f7e22c7 100644 --- a/src/gateway/test-helpers.plugin-registry.ts +++ b/src/gateway/test-helpers.plugin-registry.ts @@ -28,16 +28,21 @@ const pluginRegistryState = resolveGlobalSingleton(GATEWAY_TEST_PLUGIN_REGISTRY_ setActivePluginRegistry(pluginRegistryState.registry); +function replaceTestPluginRegistry(registry: PluginRegistry): void { + // Gateway requests retain the startup registry object. Update that owned + // snapshot in place so per-test fixtures exercise the same request scope. + Object.assign(pluginRegistryState.registry, registry); + setActivePluginRegistry(pluginRegistryState.registry); +} + /** Installs a plugin registry fixture as the active runtime registry. */ export function setTestPluginRegistry(registry: PluginRegistry): void { - pluginRegistryState.registry = registry; - setActivePluginRegistry(registry); + replaceTestPluginRegistry(registry); } /** Restores the default empty gateway test plugin registry. */ export function resetTestPluginRegistry(): void { - pluginRegistryState.registry = createStubPluginRegistry(); - setActivePluginRegistry(pluginRegistryState.registry); + replaceTestPluginRegistry(createStubPluginRegistry()); } /** Returns the currently active gateway test plugin registry. */ diff --git a/src/infra/outbound/deliver.queue-adapter-availability.test.ts b/src/infra/outbound/deliver.queue-adapter-availability.test.ts new file mode 100644 index 000000000000..197b46f76255 --- /dev/null +++ b/src/infra/outbound/deliver.queue-adapter-availability.test.ts @@ -0,0 +1,178 @@ +import { expectDefined } from "@openclaw/normalization-core"; +import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { createDefaultDeps } from "../../cli/deps.js"; +import type { OpenClawConfig } from "../../config/config.js"; +import type { PluginRegistry } from "../../plugins/registry-types.js"; +import { createEmptyPluginRegistry } from "../../plugins/registry.js"; +import { resetPluginRuntimeStateForTest, setActivePluginRegistry } from "../../plugins/runtime.js"; +import { withPluginRuntimeRegistryScope } from "../../plugins/runtime/gateway-request-scope.js"; +import { createOutboundTestPlugin, createTestRegistry } from "../../test-utils/channel-plugins.js"; +import { getDeliveryQueueEntryStatus } from "../delivery-queue-sqlite.js"; +import { + boundedCronCompletionRetention, + matrixOutboundForQueueTest, +} from "./deliver.queue-integration.test-support.js"; +import { OUTBOUND_DELIVERY_QUEUE_NAME } from "./delivery-queue-media-staging.js"; +import { loadPendingDeliveries } from "./delivery-queue-storage.js"; +import { recoverPendingDeliveries, type DeliverFn } from "./delivery-queue.js"; +import { + createRecoveryLog, + installDeliveryQueueTmpDirHooks, + setQueuedEntryState, +} from "./delivery-queue.test-helpers.js"; + +let deliverOutboundPayloads: typeof import("./deliver.js").deliverOutboundPayloads; + +type RuntimeSender = ( + to: string, + text: string, + options?: Record, +) => Promise; + +describe("queued lazy outbound adapter availability", () => { + const fixtures = installDeliveryQueueTmpDirHooks(); + let tmpDir: string; + + beforeAll(async () => { + ({ deliverOutboundPayloads } = await import("./deliver.js")); + }); + + beforeEach(() => { + tmpDir = fixtures.tmpDir(); + }); + + afterEach(() => { + resetPluginRuntimeStateForTest(); + setActivePluginRegistry(createEmptyPluginRegistry()); + }); + + it("retries adapter lookup failures without preserving false send evidence", async () => { + process.env.OPENCLAW_STATE_DIR = tmpDir; + const emptyRegistry = createEmptyPluginRegistry(); + const outerRegistry = createTestRegistry([ + { + pluginId: "matrix", + source: "test-outer", + plugin: createOutboundTestPlugin({ id: "matrix", outbound: matrixOutboundForQueueTest }), + }, + ]); + const restoredRegistry = createTestRegistry([ + { + pluginId: "matrix", + source: "test-restored", + plugin: createOutboundTestPlugin({ + id: "matrix", + outbound: { + deliveryMode: "direct", + sendText: async () => ({ channel: "matrix", messageId: "matrix-recovered" }), + }, + }), + }, + ]); + setActivePluginRegistry(outerRegistry); + const defaultDeps = createDefaultDeps() as Record; + const lazyRuntimeSender = expectDefined(defaultDeps.matrix, "matrix runtime sender"); + let scopedRuntimeRegistry: PluginRegistry = emptyRegistry; + const deps = { + matrix: (...args: Parameters) => + withPluginRuntimeRegistryScope(scopedRuntimeRegistry, async () => { + setActivePluginRegistry(emptyRegistry); + return await lazyRuntimeSender(...args); + }), + }; + const deliveryIntentId = "cron-direct-delivery:v1:lazy-adapter-recovery"; + const params = { + cfg: {} as OpenClawConfig, + channel: "matrix" as const, + to: "!room:example", + payloads: [{ text: "recover after adapter registration" }], + deps, + queuePolicy: "required" as const, + deliveryIntentId, + completionRetention: boundedCronCompletionRetention, + maxRetries: 2, + reusePendingDeliveryIntent: true, + }; + + await expect(deliverOutboundPayloads(params)).rejects.toThrow( + "matrix outbound adapter is unavailable.", + ); + const initialEntry = expectDefined( + (await loadPendingDeliveries(tmpDir))[0], + "initial queued delivery", + ); + expect(initialEntry).toMatchObject({ + id: deliveryIntentId, + retryCount: 1, + }); + expect(initialEntry.recoveryState).toBeUndefined(); + expect(initialEntry.platformSendStartedAt).toBeUndefined(); + + setQueuedEntryState(tmpDir, deliveryIntentId, { + retryCount: initialEntry.retryCount, + lastAttemptAt: 1, + lastError: initialEntry.lastError, + }); + scopedRuntimeRegistry = restoredRegistry; + setActivePluginRegistry(outerRegistry); + const recoveryDeliver = vi.fn(async (deliveryParams) => + deliverOutboundPayloads({ ...deliveryParams, deps }), + ); + + await recoverPendingDeliveries({ + cfg: {} as OpenClawConfig, + deliver: recoveryDeliver, + log: createRecoveryLog(), + stateDir: tmpDir, + }); + + expect(recoveryDeliver).toHaveBeenCalledOnce(); + expect( + getDeliveryQueueEntryStatus(OUTBOUND_DELIVERY_QUEUE_NAME, deliveryIntentId, tmpDir), + ).toBe("completed"); + }); + + it("does not replay a provider call that already crossed the ambiguous send boundary", async () => { + process.env.OPENCLAW_STATE_DIR = tmpDir; + setActivePluginRegistry( + createTestRegistry([ + { + pluginId: "matrix", + source: "test-ambiguous", + plugin: createOutboundTestPlugin({ id: "matrix", outbound: matrixOutboundForQueueTest }), + }, + ]), + ); + const sendMatrix = vi.fn().mockRejectedValue(new Error("provider result was lost")); + const deliveryIntentId = "cron-direct-delivery:v1:ambiguous-adapter-result"; + + await expect( + deliverOutboundPayloads({ + cfg: {} as OpenClawConfig, + channel: "matrix", + to: "!room:example", + payloads: [{ text: "ambiguous send" }], + deps: { matrix: sendMatrix }, + queuePolicy: "required", + deliveryIntentId, + completionRetention: boundedCronCompletionRetention, + reusePendingDeliveryIntent: true, + }), + ).rejects.toThrow("provider result was lost"); + expect((await loadPendingDeliveries(tmpDir))[0]).toMatchObject({ + id: deliveryIntentId, + recoveryState: "send_attempt_started", + }); + + const recoveryDeliver = vi.fn(async () => []); + await recoverPendingDeliveries({ + cfg: {} as OpenClawConfig, + deliver: recoveryDeliver, + log: createRecoveryLog(), + stateDir: tmpDir, + }); + + expect(recoveryDeliver).not.toHaveBeenCalled(); + expect(sendMatrix).toHaveBeenCalledOnce(); + }); +});