From 07b7d6446ca14a706787111b6cf075eb34353af6 Mon Sep 17 00:00:00 2001 From: joshavant <830519+joshavant@users.noreply.github.com> Date: Fri, 31 Jul 2026 23:46:06 -0500 Subject: [PATCH] fix(matrix): reconcile durable sends after response loss --- docs/docs_map.md | 1 + docs/plugins/sdk-channel-outbound.md | 31 ++ extensions/matrix/src/actions.ts | 31 +- .../matrix/src/channel-message-adapter.ts | 62 +++ .../src/channel.message-adapter.test.ts | 62 +++ extensions/matrix/src/channel.runtime.ts | 3 + extensions/matrix/src/channel.ts | 28 +- .../matrix/src/matrix/delivery-plan.test.ts | 358 ++++++++++++ extensions/matrix/src/matrix/delivery-plan.ts | 510 ++++++++++++++++++ extensions/matrix/src/matrix/sdk.test.ts | 76 +++ .../matrix/src/matrix/sdk/client-base.ts | 86 ++- .../matrix/src/matrix/sdk/client-core.ts | 103 +++- extensions/matrix/src/matrix/send.test.ts | 94 +++- extensions/matrix/src/matrix/send.ts | 311 ++++++----- extensions/matrix/src/matrix/send/types.ts | 8 + extensions/matrix/src/outbound.ts | 16 + extensions/matrix/src/test-runtime.ts | 13 +- src/channels/message/types.ts | 6 + src/channels/plugins/outbound.types.ts | 2 + .../channel-outbound-send.test.ts | 4 + src/cli/send-runtime/channel-outbound-send.ts | 6 + ...at-runner.ack-token-heartbeat-acks.test.ts | 3 + src/infra/outbound/deliver-channel.ts | 7 +- src/infra/outbound/deliver-contracts.ts | 2 +- src/infra/outbound/deliver-queue.ts | 26 +- src/infra/outbound/deliver.test.ts | 73 ++- .../outbound/delivery-queue-reconciliation.ts | 52 +- src/infra/outbound/delivery-queue-recovery.ts | 107 +++- .../outbound/delivery-queue.recovery.test.ts | 8 + src/infra/outbound/message-plan.test.ts | 12 +- src/infra/outbound/message-plan.ts | 30 +- 31 files changed, 1898 insertions(+), 233 deletions(-) create mode 100644 extensions/matrix/src/channel-message-adapter.ts create mode 100644 extensions/matrix/src/matrix/delivery-plan.test.ts create mode 100644 extensions/matrix/src/matrix/delivery-plan.ts diff --git a/docs/docs_map.md b/docs/docs_map.md index 894f6c8c48d2..c90ac0ed6ad0 100644 --- a/docs/docs_map.md +++ b/docs/docs_map.md @@ -7693,6 +7693,7 @@ Do not edit it by hand; run `pnpm docs:map:gen`. - H2: Delivery Evidence - H2: Existing outbound adapters - H2: Durable sends + - H3: Automatic unknown-send reconciliation - H2: Deferred delivery admission - H2: Compatibility dispatch diff --git a/docs/plugins/sdk-channel-outbound.md b/docs/plugins/sdk-channel-outbound.md index 457cd1abd729..328962205dcc 100644 --- a/docs/plugins/sdk-channel-outbound.md +++ b/docs/plugins/sdk-channel-outbound.md @@ -201,6 +201,37 @@ Use `payloadOutcomes` when a batch mixes sent, suppressed, and failed payloads. Do not infer hook cancellation from an empty legacy direct-delivery result. +### Automatic unknown-send reconciliation + +Set `message.durableFinal.automaticUnknownSendReconciliation` only when the +plugin can reconcile an ambiguous provider send from persisted, post-policy +state without rerunning modifying hooks or regenerating provider payloads. +Core considers this opt-in after hooks and cancellation, and only for exactly +one accepted prepared payload. Multi-payload batches do not opt in +automatically. + +The adapter must also advertise `capabilities.reconcileUnknownSend: true` and +provide `reconcileUnknownSend(...)`. Use `reconcileUnknownSendKinds` to name +the concrete transport branches the plugin can prove, such as `text` or +`media`. If the kind map is present, the selected branch must be `true`. +Omitting the map means the callback claims every selected branch, so prefer an +explicit map for new plugins. + +The callback must use provider-owned idempotency or authoritative readback to +return `sent` with the actual provider receipt, `not_sent` only when a fresh +send is provably safe, or `unresolved` when neither outcome can be proven. +When reconciliation is explicitly required, unsupported prepared shapes fail +before provider I/O. During recovery, missing, incomplete, or mismatched +provider proof must fail closed rather than replaying content that could +already be visible. + +If reconciliation needs provider-owned persisted evidence, implement +`afterUnknownSendTerminal(...)`. Core calls it after the ambiguous queue row +has authoritatively moved to failed, including retry-budget exhaustion. Use it +to remove provider-owned plans or payloads that are no longer needed. Cleanup +is best effort and must be idempotent; a failure is logged without making the +terminal queue row replayable again. + ## Deferred delivery admission Use `message.durableFinal.admitDeferredDelivery(...)` when a resolved account diff --git a/extensions/matrix/src/actions.ts b/extensions/matrix/src/actions.ts index 4922a6fd9b2a..cde69547f9bd 100644 --- a/extensions/matrix/src/actions.ts +++ b/extensions/matrix/src/actions.ts @@ -117,17 +117,22 @@ function buildMatrixProfileToolSchema(): NonNullable { const resolvedCfg = cfg as CoreConfig; - if (!accountId && requiresExplicitMatrixDefaultAccount(resolvedCfg)) { - return { actions: [], capabilities: [] }; - } - const account = resolveMatrixAccount({ - cfg: resolvedCfg, - accountId: accountId ?? resolveDefaultMatrixAccountId(resolvedCfg), - }); - if (!account.enabled || !account.configured) { + const account = resolveMatrixActionAccount({ cfg: resolvedCfg, accountId }); + if (!account) { return { actions: [], capabilities: [] }; } const gate = createActionGate(account.config.actions); @@ -150,6 +155,16 @@ export const matrixMessageActions: ChannelMessageActionAdapter = { extractToolSend: ({ args }) => { return extractToolSend(args, "sendMessage"); }, + prepareSendPayload: ({ ctx, payload }) => { + if (ctx.action !== "send") { + return null; + } + const account = resolveMatrixActionAccount({ + cfg: ctx.cfg as CoreConfig, + accountId: ctx.accountId, + }); + return account && createActionGate(account.config.actions)("messages") ? payload : null; + }, handleAction: async (ctx: ChannelMessageActionContext) => { const { handleMatrixAction } = await import("./tool-actions.runtime.js"); const { action, params, cfg, accountId, mediaLocalRoots } = ctx; diff --git a/extensions/matrix/src/channel-message-adapter.ts b/extensions/matrix/src/channel-message-adapter.ts new file mode 100644 index 000000000000..3ff260675a43 --- /dev/null +++ b/extensions/matrix/src/channel-message-adapter.ts @@ -0,0 +1,62 @@ +import { createChannelMessageAdapterFromOutbound } from "openclaw/plugin-sdk/channel-outbound"; +import type { ChannelOutboundAdapter } from "openclaw/plugin-sdk/channel-send-result"; +import type { matrixChannelRuntime } from "./channel.runtime.js"; + +export function createMatrixMessageAdapter(params: { + outbound: ChannelOutboundAdapter; + getRuntime: () => Promise; +}) { + const base = createChannelMessageAdapterFromOutbound({ + id: "matrix", + outbound: params.outbound, + live: { + capabilities: { + draftPreview: true, + previewFinalization: true, + progressUpdates: true, + quietFinalization: true, + }, + finalizer: { + capabilities: { + finalEdit: true, + normalFallback: true, + discardPending: true, + previewReceipt: true, + }, + }, + }, + }); + + return { + ...base, + durableFinal: { + ...base.durableFinal, + automaticUnknownSendReconciliation: true, + capabilities: { + ...base.durableFinal?.capabilities, + afterCommit: true, + reconcileUnknownSend: true, + }, + reconcileUnknownSendKinds: { text: true, media: true }, + reconcileUnknownSend: async (ctx) => + await (await params.getRuntime()).reconcileMatrixUnknownSend(ctx), + afterUnknownSendTerminal: async (ctx) => + await (await params.getRuntime()).cleanupMatrixDeliveryPlans({ queueId: ctx.queueId }), + }, + send: { + ...base.send, + lifecycle: { + afterCommit: async (ctx) => { + if (!ctx.deliveryQueueId) { + return; + } + await ( + await params.getRuntime() + ).cleanupMatrixDeliveryPlans({ + queueId: ctx.deliveryQueueId, + }); + }, + }, + }, + } satisfies typeof base; +} diff --git a/extensions/matrix/src/channel.message-adapter.test.ts b/extensions/matrix/src/channel.message-adapter.test.ts index 3541cb8445dd..d4bcac1f15ee 100644 --- a/extensions/matrix/src/channel.message-adapter.test.ts +++ b/extensions/matrix/src/channel.message-adapter.test.ts @@ -21,6 +21,7 @@ vi.mock("./matrix/send.js", () => ({ })); vi.mock("./runtime.js", () => ({ + getOptionalMatrixRuntime: () => undefined, getMatrixRuntime: () => ({ channel: { text: { @@ -35,6 +36,8 @@ import { matrixPlugin } from "./channel.js"; const cfg = { channels: { matrix: { + homeserver: "https://matrix.example.org", + userId: "@bot:example.org", accessToken: "resolved-token", }, }, @@ -68,6 +71,58 @@ describe("matrix channel message adapter", () => { expect(matrixPlugin.meta.markdownCapable).toBe(true); }); + it("opts ordinary durable text and media sends into Matrix reconciliation", () => { + expect(matrixPlugin.message?.durableFinal).toMatchObject({ + automaticUnknownSendReconciliation: true, + capabilities: { + text: true, + media: true, + afterCommit: true, + reconcileUnknownSend: true, + }, + reconcileUnknownSendKinds: { text: true, media: true }, + }); + expect(matrixPlugin.message?.durableFinal?.capabilities?.payload).not.toBe(true); + expect(matrixPlugin.message?.durableFinal?.capabilities?.batch).not.toBe(true); + }); + + it("forwards the exact durable part topology into Matrix sends", async () => { + const sendText = matrixPlugin.message?.send?.text; + if (!sendText) { + throw new Error("Expected Matrix message adapter text sender"); + } + await sendText({ + cfg, + to: "room:!room:example", + text: "durable", + accountId: "default", + deliveryQueueId: "queue-1", + deliveryPartIndex: 2, + deliveryPartCount: 3, + }); + + expect(lastMatrixSendOptions()).toMatchObject({ + deliveryQueueId: "queue-1", + deliveryPartIndex: 2, + deliveryPartCount: 3, + }); + }); + + it("routes the standard Matrix send action through canonical durable delivery", async () => { + const prepareSendPayload = matrixPlugin.actions?.prepareSendPayload; + if (!prepareSendPayload) { + throw new Error("Expected Matrix prepared-send adapter"); + } + const payload = { text: "durable tool send" }; + + expect(prepareSendPayload({ ctx: { action: "send", cfg } as never, payload } as never)).toBe( + payload, + ); + expect( + prepareSendPayload({ ctx: { action: "edit", cfg } as never, payload } as never), + ).toBeNull(); + }); + it.each([ { name: "the current room with reply quoting disabled", @@ -224,6 +279,13 @@ describe("matrix channel message adapter", () => { messageSendingHooks: () => { expect(adapter.send?.text).toBeTypeOf("function"); }, + afterCommit: () => { + expect(adapter.send?.lifecycle?.afterCommit).toBeTypeOf("function"); + }, + reconcileUnknownSend: () => { + expect(adapter.durableFinal?.reconcileUnknownSend).toBeTypeOf("function"); + expect(adapter.durableFinal?.afterUnknownSendTerminal).toBeTypeOf("function"); + }, }, }); }); diff --git a/extensions/matrix/src/channel.runtime.ts b/extensions/matrix/src/channel.runtime.ts index ab3c08100543..9566d2dd0172 100644 --- a/extensions/matrix/src/channel.runtime.ts +++ b/extensions/matrix/src/channel.runtime.ts @@ -1,18 +1,21 @@ // Matrix plugin module implements channel behavior. import { listMatrixDirectoryGroupsLive, listMatrixDirectoryPeersLive } from "./directory-live.js"; import { resolveMatrixAuth } from "./matrix/client.js"; +import { cleanupMatrixDeliveryPlans, reconcileMatrixUnknownSend } from "./matrix/delivery-plan.js"; import { probeMatrix } from "./matrix/probe.js"; import { sendMessageMatrix, sendTypingMatrix } from "./matrix/send.js"; import { matrixOutbound } from "./outbound.js"; import { resolveMatrixTargets } from "./resolve-targets.js"; export const matrixChannelRuntime = { + cleanupMatrixDeliveryPlans, listMatrixDirectoryGroupsLive, listMatrixDirectoryPeersLive, matrixOutbound, probeMatrix, resolveMatrixAuth, resolveMatrixTargets, + reconcileMatrixUnknownSend, sendMessageMatrix, sendTypingMatrix, }; diff --git a/extensions/matrix/src/channel.ts b/extensions/matrix/src/channel.ts index fdb5e587ddf3..094cae4ba380 100644 --- a/extensions/matrix/src/channel.ts +++ b/extensions/matrix/src/channel.ts @@ -8,10 +8,7 @@ import type { ChannelThreadingToolContext, } from "openclaw/plugin-sdk/channel-contract"; import { createChatChannelPlugin, type ChannelPlugin } from "openclaw/plugin-sdk/channel-core"; -import { - createChannelMessageAdapterFromOutbound, - createRuntimeOutboundDelegates, -} from "openclaw/plugin-sdk/channel-outbound"; +import { createRuntimeOutboundDelegates } from "openclaw/plugin-sdk/channel-outbound"; import { createAllowlistProviderOpenWarningCollector, projectAccountConfigWarningCollector, @@ -44,6 +41,7 @@ import { import { matrixMessageActions } from "./actions.js"; import { matrixApprovalCapability } from "./approval-native.js"; import { createMatrixPairingText, createMatrixProbeAccount } from "./channel-account-paths.js"; +import { createMatrixMessageAdapter } from "./channel-message-adapter.js"; import { matrixPluginBase } from "./channel.setup.js"; import { DEFAULT_ACCOUNT_ID } from "./config-adapter.js"; import { @@ -346,6 +344,8 @@ const matrixChannelOutbound: ChannelOutboundAdapter = { replyTo: true, thread: true, messageSendingHooks: true, + afterCommit: true, + reconcileUnknownSend: true, }, }, presentationCapabilities: { @@ -392,25 +392,9 @@ const matrixChannelOutbound: ChannelOutboundAdapter = { }), }; -const matrixMessageAdapter = createChannelMessageAdapterFromOutbound({ - id: "matrix", +const matrixMessageAdapter = createMatrixMessageAdapter({ outbound: matrixChannelOutbound, - live: { - capabilities: { - draftPreview: true, - previewFinalization: true, - progressUpdates: true, - quietFinalization: true, - }, - finalizer: { - capabilities: { - finalEdit: true, - normalFallback: true, - discardPending: true, - previewReceipt: true, - }, - }, - }, + getRuntime: loadMatrixChannelRuntime, }); export const matrixPlugin: ChannelPlugin = diff --git a/extensions/matrix/src/matrix/delivery-plan.test.ts b/extensions/matrix/src/matrix/delivery-plan.test.ts new file mode 100644 index 000000000000..0210a277a5d3 --- /dev/null +++ b/extensions/matrix/src/matrix/delivery-plan.test.ts @@ -0,0 +1,358 @@ +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { + resetPluginBlobStoreForTests, + resetPluginStateStoreForTests, +} from "openclaw/plugin-sdk/plugin-state-test-runtime"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { installMatrixTestRuntime } from "../test-runtime.js"; +import { + cleanupMatrixDeliveryPlans, + createMatrixPlannedEvents, + loadMatrixDeliveryPlan, + persistMatrixDeliveryPlan, + reconcileMatrixUnknownSend, + resolveMatrixDurableDeliveryIdentity, +} from "./delivery-plan.js"; + +const client = { + getTransactionScopeId: vi.fn(async () => "scope-1"), + getMessageWireEventType: vi.fn(async () => "m.room.message" as const), + sendMessage: vi.fn( + async ( + roomId: string, + _content: unknown, + transactionId?: string, + beforeWireDispatch?: (dispatch: { + roomId: string; + eventType: "m.room.message"; + transactionId: string; + requestPath: string; + }) => Promise, + ) => { + const resolvedTransactionId = transactionId ?? "missing"; + await beforeWireDispatch?.({ + roomId, + eventType: "m.room.message", + transactionId: resolvedTransactionId, + requestPath: `/_matrix/client/v3/rooms/${encodeURIComponent(roomId)}/send/m.room.message/${resolvedTransactionId}`, + }); + return `$${resolvedTransactionId}`; + }, + ), +}; + +vi.mock("./send/client.js", () => ({ + withResolvedMatrixSendClient: async ( + _opts: unknown, + run: (resolved: typeof client) => Promise, + ) => await run(client), +})); + +vi.mock("./send/targets.js", () => ({ + resolveMatrixRoomId: vi.fn(async () => "!room:example.org"), +})); + +let stateDir = ""; + +function identity(queueId = "queue-1", partIndex = 0, partCount = 1) { + const resolved = resolveMatrixDurableDeliveryIdentity({ queueId, partIndex, partCount }); + if (!resolved) { + throw new Error("expected durable Matrix identity"); + } + return resolved; +} + +function events(deliveryIdentity = identity()) { + return createMatrixPlannedEvents({ + identity: deliveryIdentity, + events: [ + { + receiptKind: "text", + content: { msgtype: "m.text", body: "durable hello" }, + }, + ], + }); +} + +async function persist( + params: { + queueId?: string; + partIndex?: number; + partCount?: number; + accountId?: string; + scope?: string; + } = {}, +) { + const deliveryIdentity = identity(params.queueId, params.partIndex, params.partCount); + const plannedEvents = events(deliveryIdentity); + return await persistMatrixDeliveryPlan({ + identity: deliveryIdentity, + accountId: params.accountId ?? "default", + roomId: "!room:example.org", + transactionScopeId: params.scope ?? "scope-1", + wireEventType: "m.room.message", + events: plannedEvents, + dispatch: { + roomId: "!room:example.org", + eventType: "m.room.message", + transactionId: plannedEvents[0]!.transactionId, + requestPath: `/_matrix/client/v3/rooms/!room%3Aexample.org/send/m.room.message/${plannedEvents[0]!.transactionId}`, + }, + }); +} + +function reconciliationContext(queueId = "queue-1") { + return { + cfg: {}, + queueId, + channel: "matrix", + to: "room:!room:example.org", + accountId: "default", + enqueuedAt: 1, + payloads: [{ text: "durable hello" }], + retryCount: 0, + } as const; +} + +describe("Matrix durable delivery plans", () => { + beforeEach(() => { + stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-matrix-plan-")); + installMatrixTestRuntime({ stateDir }); + client.getTransactionScopeId.mockReset().mockResolvedValue("scope-1"); + client.getMessageWireEventType.mockReset().mockResolvedValue("m.room.message"); + client.sendMessage.mockClear(); + }); + + afterEach(() => { + resetPluginBlobStoreForTests({ closeDatabase: false }); + resetPluginStateStoreForTests(); + fs.rmSync(stateDir, { recursive: true, force: true }); + }); + + it("persists one exact plan and rejects a different plan for the same queue part", async () => { + const plan = await persist(); + const deliveryIdentity = identity(); + expect(plan.events[0]).toMatchObject({ + receiptKind: "text", + content: { msgtype: "m.text", body: "durable hello" }, + }); + expect(plan.events[0]?.transactionId).toMatch(/^oc_/); + await expect( + loadMatrixDeliveryPlan({ + identity: deliveryIdentity, + accountId: "default", + roomId: "!room:example.org", + transactionScopeId: "scope-1", + wireEventType: "m.room.message", + }), + ).resolves.toEqual(plan); + + const changedEvents = createMatrixPlannedEvents({ + identity: deliveryIdentity, + events: [{ receiptKind: "text", content: { msgtype: "m.text", body: "changed" } }], + }); + await expect( + persistMatrixDeliveryPlan({ + identity: deliveryIdentity, + accountId: "default", + roomId: "!room:example.org", + transactionScopeId: "scope-1", + wireEventType: "m.room.message", + events: changedEvents, + dispatch: { + roomId: "!room:example.org", + eventType: "m.room.message", + transactionId: changedEvents[0]!.transactionId, + requestPath: `/_matrix/client/v3/rooms/!room%3Aexample.org/send/m.room.message/${changedEvents[0]!.transactionId}`, + }, + }), + ).rejects.toThrow("no longer matches the prepared event batch"); + }); + + it("reissues the exact stored event with its transaction id and reports the provider event id", async () => { + const plan = await persist(); + client.sendMessage.mockImplementationOnce( + async (roomId, _content, transactionId, beforeWireDispatch) => { + await beforeWireDispatch?.({ + roomId, + eventType: "m.room.message", + transactionId: transactionId ?? "missing", + requestPath: `/_matrix/client/v3/rooms/${encodeURIComponent(roomId)}/send/m.room.message/${transactionId}`, + }); + return "$event-1"; + }, + ); + + await expect(reconcileMatrixUnknownSend(reconciliationContext())).resolves.toMatchObject({ + status: "sent", + messageId: "$event-1", + receipt: { + primaryPlatformMessageId: "$event-1", + platformMessageIds: ["$event-1"], + parts: [{ platformMessageId: "$event-1", kind: "text" }], + }, + }); + expect(client.sendMessage).toHaveBeenCalledWith( + "!room:example.org", + plan.events[0]!.content, + plan.events[0]!.transactionId, + expect.any(Function), + ); + }); + + it("preserves ordered typed receipt parts and the final event identity", async () => { + const deliveryIdentity = identity("queue-multi-event"); + const plannedEvents = createMatrixPlannedEvents({ + identity: deliveryIdentity, + events: [ + { receiptKind: "media", content: { msgtype: "m.image", body: "caption" } }, + { receiptKind: "text", content: { msgtype: "m.text", body: "follow-up" } }, + ], + }); + await persistMatrixDeliveryPlan({ + identity: deliveryIdentity, + accountId: "default", + roomId: "!room:example.org", + transactionScopeId: "scope-1", + wireEventType: "m.room.message", + events: plannedEvents, + dispatch: { + roomId: "!room:example.org", + eventType: "m.room.message", + transactionId: plannedEvents[0]!.transactionId, + requestPath: `/_matrix/client/v3/rooms/!room%3Aexample.org/send/m.room.message/${plannedEvents[0]!.transactionId}`, + }, + }); + client.sendMessage.mockResolvedValueOnce("$media-event").mockResolvedValueOnce("$text-event"); + + await expect( + reconcileMatrixUnknownSend({ + ...reconciliationContext("queue-multi-event"), + effectiveReplyToId: "$reply", + threadId: "$thread", + }), + ).resolves.toMatchObject({ + status: "sent", + messageId: "$text-event", + receipt: { + primaryPlatformMessageId: "$media-event", + platformMessageIds: ["$media-event", "$text-event"], + replyToId: "$reply", + threadId: "$thread", + parts: [ + { + platformMessageId: "$media-event", + kind: "media", + index: 0, + replyToId: "$reply", + threadId: "$thread", + }, + { + platformMessageId: "$text-event", + kind: "text", + index: 1, + replyToId: "$reply", + threadId: "$thread", + }, + ], + }, + }); + }); + + it("fails closed without provider I/O when any expected part plan is missing", async () => { + const incompleteIdentity = identity("queue-incomplete", 0, 2); + await persist({ queueId: "queue-incomplete", partIndex: 0, partCount: 2 }); + + await expect( + reconcileMatrixUnknownSend(reconciliationContext("queue-incomplete")), + ).resolves.toMatchObject({ + status: "unresolved", + retryable: false, + error: expect.stringContaining("incomplete event plan"), + }); + expect(client.sendMessage).not.toHaveBeenCalled(); + await expect( + loadMatrixDeliveryPlan({ + identity: incompleteIdentity, + accountId: "default", + roomId: "!room:example.org", + transactionScopeId: "scope-1", + wireEventType: "m.room.message", + }), + ).resolves.toBeNull(); + }); + + it("fails closed when the active transaction scope differs", async () => { + const scopeIdentity = identity("queue-scope"); + await persist({ queueId: "queue-scope", scope: "old-scope" }); + + await expect( + reconcileMatrixUnknownSend(reconciliationContext("queue-scope")), + ).resolves.toMatchObject({ + status: "unresolved", + retryable: false, + error: expect.stringContaining("no longer matches the active delivery target"), + }); + expect(client.sendMessage).not.toHaveBeenCalled(); + await expect( + loadMatrixDeliveryPlan({ + identity: scopeIdentity, + accountId: "default", + roomId: "!room:example.org", + transactionScopeId: "old-scope", + wireEventType: "m.room.message", + }), + ).resolves.toBeNull(); + }); + + it("fails closed when the SDK selects a different Matrix endpoint path", async () => { + await persist({ queueId: "queue-route" }); + client.sendMessage.mockImplementationOnce( + async (roomId, _content, transactionId, beforeWireDispatch) => { + await beforeWireDispatch?.({ + roomId, + eventType: "m.room.message", + transactionId: transactionId ?? "missing", + requestPath: `/_matrix/client/v4/rooms/${encodeURIComponent(roomId)}/send/m.room.message/${transactionId}`, + }); + return "$must-not-send"; + }, + ); + + await expect( + reconcileMatrixUnknownSend(reconciliationContext("queue-route")), + ).resolves.toMatchObject({ + status: "unresolved", + retryable: false, + error: expect.stringContaining("no longer matches the prepared event batch"), + }); + }); + + it("removes all plans for a committed queue without touching another queue", async () => { + await persist({ queueId: "queue-clean" }); + await persist({ queueId: "queue-keep" }); + + await cleanupMatrixDeliveryPlans({ queueId: "queue-clean" }); + + await expect( + loadMatrixDeliveryPlan({ + identity: identity("queue-clean"), + accountId: "default", + roomId: "!room:example.org", + transactionScopeId: "scope-1", + wireEventType: "m.room.message", + }), + ).resolves.toBeNull(); + await expect( + loadMatrixDeliveryPlan({ + identity: identity("queue-keep"), + accountId: "default", + roomId: "!room:example.org", + transactionScopeId: "scope-1", + wireEventType: "m.room.message", + }), + ).resolves.not.toBeNull(); + }); +}); diff --git a/extensions/matrix/src/matrix/delivery-plan.ts b/extensions/matrix/src/matrix/delivery-plan.ts new file mode 100644 index 000000000000..8b5940bb3ff5 --- /dev/null +++ b/extensions/matrix/src/matrix/delivery-plan.ts @@ -0,0 +1,510 @@ +// Matrix-owned event plans reconcile ambiguous sends through native transaction idempotency. +import { createHash } from "node:crypto"; +import type { + ChannelMessageUnknownSendContext, + ChannelMessageUnknownSendReconciliationResult, + MessageReceipt, + MessageReceiptPartKind, +} from "openclaw/plugin-sdk/channel-outbound"; +import { getMatrixRuntime } from "../runtime.js"; +import type { MatrixClient } from "./sdk.js"; +import type { MatrixMessageWireDispatch } from "./sdk/client-base.js"; +import { withResolvedMatrixSendClient } from "./send/client.js"; +import { resolveMatrixRoomId } from "./send/targets.js"; +import type { MatrixOutboundContent } from "./send/types.js"; + +const DELIVERY_PLAN_VERSION = 1; +const DELIVERY_PLAN_NAMESPACE = "outbound-delivery-plans"; +// Recovery exhausts its normal retry schedule within minutes. Keep a one-day +// interruption cushion without retaining terminal message content for a year. +const DELIVERY_PLAN_TTL_MS = 24 * 60 * 60 * 1000; + +class MatrixDeliveryPlanInvariantError extends Error { + constructor(message: string) { + super(message); + this.name = "MatrixDeliveryPlanInvariantError"; + } +} + +export type MatrixPreparedEvent = { + transactionId: string; + receiptKind: MessageReceiptPartKind; + content: MatrixOutboundContent; +}; + +type MatrixDeliveryIdentity = { + queueId: string; + partIndex: number; + partCount: number; +}; + +type MatrixDeliveryPlan = { + version: typeof DELIVERY_PLAN_VERSION; + queueId: string; + accountId: string; + roomId: string; + wireEventType: "m.room.message" | "m.room.encrypted"; + endpointPrefix: string; + transactionScopeId: string; + partIndex: number; + partCount: number; + events: MatrixPreparedEvent[]; +}; + +function createDeliveryPlanStore() { + return getMatrixRuntime().state.openBlobStore>({ + namespace: DELIVERY_PLAN_NAMESPACE, + maxEntries: 10_000, + maxBytesPerEntry: 8 * 1024 * 1024, + maxBytesPerNamespace: 256 * 1024 * 1024, + overflowPolicy: "reject-new", + defaultTtlMs: DELIVERY_PLAN_TTL_MS, + }); +} + +function requireIndex(value: number, label: string): number { + if (!Number.isSafeInteger(value) || value < 0) { + throw new Error(`Matrix durable delivery ${label} must be a non-negative integer`); + } + return value; +} + +function requirePartCount(value: number | undefined): number { + if (!Number.isSafeInteger(value) || (value ?? 0) < 1) { + throw new Error("Matrix durable delivery part count must be a positive integer"); + } + return value!; +} + +function queuePrefix(queueId: string): string { + const normalized = queueId.trim(); + if (!normalized) { + throw new Error("Matrix durable delivery requires a queue id"); + } + return `${createHash("sha256").update(normalized).digest("hex")}.`; +} + +function planKey(identity: MatrixDeliveryIdentity): string { + return `${queuePrefix(identity.queueId)}${requireIndex(identity.partIndex, "part index")}`; +} + +function transactionId(identity: MatrixDeliveryIdentity, eventIndex: number): string { + const digest = createHash("sha256") + .update(identity.queueId) + .update("\0") + .update(String(requireIndex(identity.partIndex, "part index"))) + .update("\0") + .update(String(requireIndex(eventIndex, "event index"))) + .digest("base64url"); + return `oc_${digest}`; +} + +const RECEIPT_KINDS = new Set([ + "text", + "media", + "voice", + "poll", + "card", + "preview", + "unknown", +]); + +function isPlan(value: unknown): value is MatrixDeliveryPlan { + if (!value || typeof value !== "object") { + return false; + } + const plan = value as Partial; + return ( + plan.version === DELIVERY_PLAN_VERSION && + typeof plan.queueId === "string" && + Boolean(plan.queueId.trim()) && + typeof plan.accountId === "string" && + typeof plan.roomId === "string" && + Boolean(plan.roomId.trim()) && + (plan.wireEventType === "m.room.message" || plan.wireEventType === "m.room.encrypted") && + typeof plan.endpointPrefix === "string" && + Boolean(plan.endpointPrefix.trim()) && + typeof plan.transactionScopeId === "string" && + Boolean(plan.transactionScopeId.trim()) && + Number.isSafeInteger(plan.partIndex) && + (plan.partIndex ?? -1) >= 0 && + Number.isSafeInteger(plan.partCount) && + (plan.partCount ?? 0) > 0 && + (plan.partIndex ?? -1) < (plan.partCount ?? 0) && + Array.isArray(plan.events) && + plan.events.length > 0 && + plan.events.every( + (event) => + event && + typeof event === "object" && + typeof event.transactionId === "string" && + Boolean(event.transactionId.trim()) && + RECEIPT_KINDS.has(event.receiptKind) && + Boolean(event.content) && + typeof event.content === "object", + ) + ); +} + +function decodePlan(bytes: Uint8Array): MatrixDeliveryPlan { + let value: unknown; + try { + value = JSON.parse(new TextDecoder().decode(bytes)); + } catch { + throw new MatrixDeliveryPlanInvariantError("Matrix durable delivery plan is invalid JSON"); + } + if (!isPlan(value)) { + throw new MatrixDeliveryPlanInvariantError("Matrix durable delivery plan is invalid"); + } + return value; +} + +function assertPlanIdentity( + plan: MatrixDeliveryPlan, + params: { + identity: MatrixDeliveryIdentity; + accountId?: string | null; + roomId: string; + transactionScopeId: string; + wireEventType: "m.room.message" | "m.room.encrypted"; + }, +): void { + if ( + plan.queueId !== params.identity.queueId || + plan.partIndex !== params.identity.partIndex || + plan.partCount !== params.identity.partCount || + plan.accountId !== (params.accountId ?? "") || + plan.roomId !== params.roomId || + plan.transactionScopeId !== params.transactionScopeId || + plan.wireEventType !== params.wireEventType + ) { + throw new MatrixDeliveryPlanInvariantError( + "Matrix durable delivery plan no longer matches the active delivery target", + ); + } +} + +function endpointPrefix(dispatch: MatrixMessageWireDispatch): string { + const encodedTransactionId = encodeURIComponent(dispatch.transactionId); + if (!dispatch.requestPath.endsWith(encodedTransactionId)) { + throw new MatrixDeliveryPlanInvariantError( + "Matrix durable delivery transaction does not match its request path", + ); + } + return dispatch.requestPath.slice(0, -encodedTransactionId.length); +} + +export function createMatrixPlannedEvents(params: { + identity: MatrixDeliveryIdentity; + events: readonly Omit[]; +}): MatrixPreparedEvent[] { + return params.events.map((event, index) => ({ + ...structuredClone(event), + transactionId: transactionId(params.identity, index), + })); +} + +export function resolveMatrixDurableDeliveryIdentity(params: { + queueId?: string; + partIndex?: number; + partCount?: number; +}): MatrixDeliveryIdentity | null { + if (params.queueId === undefined) { + return null; + } + if (params.partIndex === undefined || params.partCount === undefined) { + throw new Error("Matrix durable delivery requires stable part topology"); + } + const partIndex = requireIndex(params.partIndex, "part index"); + const partCount = requirePartCount(params.partCount); + if (partIndex >= partCount) { + throw new Error("Matrix durable delivery part index must be below the part count"); + } + return { + queueId: params.queueId, + partIndex, + partCount, + }; +} + +export async function loadMatrixDeliveryPlan(params: { + identity: MatrixDeliveryIdentity; + accountId?: string | null; + roomId: string; + transactionScopeId: string; + wireEventType: "m.room.message" | "m.room.encrypted"; +}): Promise { + const entry = await createDeliveryPlanStore().lookup(planKey(params.identity)); + if (!entry) { + return null; + } + const plan = decodePlan(entry.bytes); + if (planKey(plan) !== planKey(params.identity)) { + throw new MatrixDeliveryPlanInvariantError("Matrix durable delivery plan key is invalid"); + } + assertPlanIdentity(plan, params); + return structuredClone(plan); +} + +export async function persistMatrixDeliveryPlan(params: { + identity: MatrixDeliveryIdentity; + accountId?: string | null; + roomId: string; + transactionScopeId: string; + wireEventType: "m.room.message" | "m.room.encrypted"; + events: readonly MatrixPreparedEvent[]; + dispatch: MatrixMessageWireDispatch; +}): Promise { + if (params.events.length === 0) { + throw new Error("Matrix durable delivery plan must contain at least one event"); + } + if ( + params.dispatch.roomId !== params.roomId || + params.dispatch.eventType !== params.wireEventType || + !params.events.some((event) => event.transactionId === params.dispatch.transactionId) + ) { + throw new MatrixDeliveryPlanInvariantError( + "Matrix durable delivery was dispatched to an unexpected endpoint", + ); + } + const partCount = requirePartCount(params.identity.partCount); + const events = params.events.map((event, index) => { + if (event.transactionId !== transactionId(params.identity, index)) { + throw new MatrixDeliveryPlanInvariantError( + "Matrix durable delivery plan has an invalid transaction identifier", + ); + } + return structuredClone(event); + }); + const plan: MatrixDeliveryPlan = { + version: DELIVERY_PLAN_VERSION, + queueId: params.identity.queueId, + accountId: params.accountId ?? "", + roomId: params.roomId, + wireEventType: params.wireEventType, + // Matrix idempotency includes the HTTP endpoint. Persist the SDK-selected + // prefix so an API-route change fails before replay reaches the homeserver. + endpointPrefix: endpointPrefix(params.dispatch), + transactionScopeId: params.transactionScopeId, + partIndex: requireIndex(params.identity.partIndex, "part index"), + partCount, + events, + }; + const store = createDeliveryPlanStore(); + await store.deleteExpired(); + const bytes = new TextEncoder().encode(JSON.stringify(plan)); + if (await store.registerIfAbsent(planKey(params.identity), bytes, {})) { + return plan; + } + const existing = await loadMatrixDeliveryPlan(params); + if (!existing || JSON.stringify(existing) !== JSON.stringify(plan)) { + throw new MatrixDeliveryPlanInvariantError( + "Matrix durable delivery plan no longer matches the prepared event batch", + ); + } + return existing; +} + +async function loadQueuePlans(queueId: string): Promise { + const store = createDeliveryPlanStore(); + const keys = (await store.entries()) + .filter((entry) => entry.key.startsWith(queuePrefix(queueId))) + .map((entry) => entry.key); + return await Promise.all( + keys.map(async (key) => { + const entry = await store.lookup(key); + if (!entry) { + throw new MatrixDeliveryPlanInvariantError( + "Matrix durable delivery plan disappeared during reconciliation", + ); + } + const plan = decodePlan(entry.bytes); + if (key !== planKey(plan)) { + throw new MatrixDeliveryPlanInvariantError("Matrix durable delivery plan key is invalid"); + } + return plan; + }), + ); +} + +function assertCompletePartTopology(plans: readonly MatrixDeliveryPlan[]): void { + const partCount = plans[0]?.partCount; + if (!partCount) { + throw new MatrixDeliveryPlanInvariantError("Matrix ambiguous delivery has no event plan"); + } + if (plans.some((plan) => plan.partCount !== partCount)) { + throw new MatrixDeliveryPlanInvariantError( + "Matrix durable delivery plan part topology is inconsistent", + ); + } + const storedParts = new Set(plans.map((plan) => plan.partIndex)); + if ( + storedParts.size !== partCount || + Array.from({ length: partCount }, (_, partIndex) => partIndex).some( + (partIndex) => !storedParts.has(partIndex), + ) + ) { + throw new MatrixDeliveryPlanInvariantError( + "Matrix ambiguous delivery has an incomplete event plan", + ); + } +} + +function createReconciledMatrixReceipt(params: { + results: readonly { eventId: string; receiptKind: MessageReceiptPartKind }[]; + replyToId?: string; + threadId?: string; +}): MessageReceipt { + const uniqueResults = params.results.filter( + (result, index, results) => + results.findIndex((entry) => entry.eventId === result.eventId) === index, + ); + const platformMessageIds = uniqueResults.map((result) => result.eventId); + return { + ...(platformMessageIds[0] ? { primaryPlatformMessageId: platformMessageIds[0] } : {}), + platformMessageIds, + parts: uniqueResults.map((result, index) => { + const part: NonNullable[number] = { + platformMessageId: result.eventId, + kind: result.receiptKind, + index, + }; + if (params.replyToId) { + part.replyToId = params.replyToId; + } + if (params.threadId) { + part.threadId = params.threadId; + } + return part; + }), + ...(params.replyToId ? { replyToId: params.replyToId } : {}), + ...(params.threadId ? { threadId: params.threadId } : {}), + sentAt: Date.now(), + }; +} + +function describeError(value: unknown): string { + if (value instanceof Error) { + return value.message; + } + return typeof value === "string" ? value : "unknown error"; +} + +async function requireTransactionScope(client: MatrixClient): Promise { + const scope = (await client.getTransactionScopeId()).trim(); + if (!scope) { + throw new MatrixDeliveryPlanInvariantError( + "Matrix durable delivery requires a stable transaction scope", + ); + } + return scope; +} + +export async function reconcileMatrixUnknownSend( + ctx: ChannelMessageUnknownSendContext, +): Promise { + try { + if (ctx.payloads.length !== 1) { + throw new MatrixDeliveryPlanInvariantError( + "Matrix reconciliation requires exactly one prepared payload", + ); + } + const plans = await loadQueuePlans(ctx.queueId); + if (plans.length === 0) { + throw new MatrixDeliveryPlanInvariantError( + "Matrix ambiguous delivery has no persisted event plan", + ); + } + assertCompletePartTopology(plans); + return await withResolvedMatrixSendClient( + { cfg: ctx.cfg, accountId: ctx.accountId }, + async (client) => { + const transactionScopeId = await requireTransactionScope(client); + const roomId = await resolveMatrixRoomId(client, ctx.to); + const wireEventType = await client.getMessageWireEventType(roomId); + const orderedPlans = [...plans].toSorted((left, right) => left.partIndex - right.partIndex); + const results: Array<{ + eventId: string; + receiptKind: MessageReceiptPartKind; + }> = []; + for (const plan of orderedPlans) { + assertPlanIdentity(plan, { + identity: plan, + accountId: ctx.accountId, + roomId, + transactionScopeId, + wireEventType, + }); + for (const event of plan.events) { + results.push({ + eventId: await client.sendMessage( + roomId, + event.content, + event.transactionId, + async (dispatch) => { + await persistMatrixDeliveryPlan({ + identity: plan, + accountId: ctx.accountId, + roomId, + transactionScopeId, + wireEventType, + events: plan.events, + dispatch, + }); + }, + ), + receiptKind: event.receiptKind, + }); + } + } + const replyToId = + ctx.effectiveReplyToId !== undefined + ? ctx.effectiveReplyToId + : ctx.replyToMode === "off" + ? undefined + : ctx.replyToId; + const threadId = ctx.threadId == null ? undefined : String(ctx.threadId); + const receipt = createReconciledMatrixReceipt({ + results, + ...(replyToId ? { replyToId } : {}), + ...(threadId ? { threadId } : {}), + }); + return { + status: "sent", + messageId: receipt.platformMessageIds.at(-1), + receipt, + }; + }, + ); + } catch (error) { + const retryable = !(error instanceof MatrixDeliveryPlanInvariantError); + let cleanupError: unknown; + if (!retryable) { + // Core terminally retires non-retryable reconciliation. Remove the plan + // here so a fail-closed Matrix verdict cannot retain payload content. + try { + await cleanupMatrixDeliveryPlans({ queueId: ctx.queueId }); + } catch (cleanupFailure) { + cleanupError = cleanupFailure; + } + } + const errorMessage = describeError(error); + return { + status: "unresolved", + error: + cleanupError === undefined + ? errorMessage + : `${errorMessage}; Matrix delivery-plan cleanup failed: ${describeError(cleanupError)}`, + retryable, + }; + } +} + +export async function cleanupMatrixDeliveryPlans(ctx: { queueId: string }): Promise { + const store = createDeliveryPlanStore(); + await store.deleteExpired(); + const keys = (await store.entries()) + .filter((entry) => entry.key.startsWith(queuePrefix(ctx.queueId))) + .map((entry) => entry.key); + await Promise.all(keys.map(async (key) => await store.delete(key))); +} diff --git a/extensions/matrix/src/matrix/sdk.test.ts b/extensions/matrix/src/matrix/sdk.test.ts index 55da3aa201de..8ad7425d54dd 100644 --- a/extensions/matrix/src/matrix/sdk.test.ts +++ b/extensions/matrix/src/matrix/sdk.test.ts @@ -392,6 +392,82 @@ describe("MatrixClient request hardening", () => { expect(matrixJsClient.getAccountData).not.toHaveBeenCalled(); }); + it("uses a conservative token-and-device-scoped transaction identity", async () => { + const first = new MatrixClient("https://matrix.example.org", "token-a", { + userId: "@bot:example.org", + deviceId: "DEVICE123", + }); + const second = new MatrixClient("https://matrix.example.org", "token-b", { + userId: "@bot:example.org", + deviceId: "DEVICE123", + }); + const whoami = { user_id: "@bot:example.org", device_id: "DEVICE123" }; + vi.spyOn(first, "doRequest").mockResolvedValue(whoami); + vi.spyOn(second, "doRequest").mockResolvedValue(whoami); + + expect(await first.getTransactionScopeId()).not.toBe(await second.getTransactionScopeId()); + await expect(first.getTransactionScopeId()).resolves.toBe(await first.getTransactionScopeId()); + }); + + it("passes stable transaction ids into matrix-js-sdk timeline sends", async () => { + const client = new MatrixClient("https://matrix.example.org", "token"); + + await expect( + client.sendMessage( + "!room:example.org", + { msgtype: "m.text", body: "hello" }, + "oc_transaction", + ), + ).resolves.toBe("$sent"); + expect(matrixJsClient.sendMessage).toHaveBeenCalledWith( + "!room:example.org", + { msgtype: "m.text", body: "hello" }, + "oc_transaction", + ); + }); + + it("runs the durable plan guard after endpoint selection and before the Matrix PUT", async () => { + const order: string[] = []; + const fetchMock = vi.fn(async () => { + order.push("put"); + return new Response(JSON.stringify({ event_id: "$sent" }), { + status: 200, + headers: { "content-type": "application/json" }, + }); + }); + stubRuntimeFetch(fetchMock as unknown as typeof fetch); + const client = new MatrixClient("http://127.0.0.1:8008", "token", { + ssrfPolicy: { allowPrivateNetwork: true }, + }); + const fetchFn = lastCreateClientOpts?.fetchFn as typeof fetch; + matrixJsClient.sendMessage = vi.fn(async (roomId, _content, transactionId) => { + await fetchFn( + `http://127.0.0.1:8008/_matrix/client/v3/rooms/${encodeURIComponent(roomId)}/send/m.room.encrypted/${transactionId}`, + { method: "PUT", body: "{}" }, + ); + return { event_id: "$sent" }; + }); + + await expect( + client.sendMessage( + "!room:example.org", + { msgtype: "m.text", body: "hello" }, + "oc_transaction", + async (dispatch) => { + order.push("guard"); + expect(dispatch).toEqual({ + roomId: "!room:example.org", + eventType: "m.room.encrypted", + transactionId: "oc_transaction", + requestPath: + "/_matrix/client/v3/rooms/!room%3Aexample.org/send/m.room.encrypted/oc_transaction", + }); + }, + ), + ).resolves.toBe("$sent"); + expect(order).toEqual(["guard", "put"]); + }); + it("blocks absolute endpoints unless explicitly allowed", async () => { const fetchMock = vi.fn(async () => { return new Response("{}", { diff --git a/extensions/matrix/src/matrix/sdk/client-base.ts b/extensions/matrix/src/matrix/sdk/client-base.ts index e246b129e074..753120810ab1 100644 --- a/extensions/matrix/src/matrix/sdk/client-base.ts +++ b/extensions/matrix/src/matrix/sdk/client-base.ts @@ -1,3 +1,4 @@ +import { createHash } from "node:crypto"; import { EventEmitter } from "node:events"; import { Filter, @@ -38,6 +39,48 @@ import type { MatrixVerificationSummary } from "./verification-manager.js"; type MatrixCryptoRuntime = typeof import("./crypto-runtime.js"); +export type MatrixMessageWireDispatch = { + roomId: string; + eventType: "m.room.message" | "m.room.encrypted"; + transactionId: string; + requestPath: string; +}; + +type MatrixMessageWireDispatchGuard = (dispatch: MatrixMessageWireDispatch) => Promise; + +function resolveMessageWireDispatch( + resource: RequestInfo | URL, + init?: RequestInit, +): MatrixMessageWireDispatch | null { + const method = ( + init?.method ?? (resource instanceof Request ? resource.method : "GET") + ).toUpperCase(); + if (method !== "PUT") { + return null; + } + const rawUrl = + typeof resource === "string" + ? resource + : resource instanceof URL + ? resource.href + : resource.url; + const segments = new URL(rawUrl).pathname.split("/").filter(Boolean); + const roomsIndex = segments.lastIndexOf("rooms"); + if (roomsIndex < 0 || segments[roomsIndex + 2] !== "send" || segments.length !== roomsIndex + 5) { + return null; + } + const eventType = decodeURIComponent(segments[roomsIndex + 3] ?? ""); + if (eventType !== "m.room.message" && eventType !== "m.room.encrypted") { + return null; + } + return { + roomId: decodeURIComponent(segments[roomsIndex + 1] ?? ""), + eventType, + transactionId: decodeURIComponent(segments[roomsIndex + 4] ?? ""), + requestPath: new URL(rawUrl).pathname, + }; +} + let loadedMatrixCryptoRuntime: MatrixCryptoRuntime | null = null; export const loadMatrixCryptoRuntime = createLazyRuntimeModule(() => @@ -93,6 +136,12 @@ export abstract class MatrixClientBase { protected stopPersistPromise: Promise | null = null; protected verificationSummaryListenerBound = false; protected currentSyncState: MatrixSyncState | null = null; + protected readonly transactionScopeHomeserver: string; + protected readonly transactionScopeAccessTokenHash: string; + protected transactionScopeDeviceId: string | null; + protected transactionScopeId: string | null = null; + protected transactionScopePromise: Promise | null = null; + private readonly messageWireDispatchGuards = new Map(); readonly dms = { update: async (): Promise => { @@ -123,6 +172,9 @@ export abstract class MatrixClientBase { dispatcherPolicy?: PinnedDispatcherPolicy; } = {}, ) { + this.transactionScopeHomeserver = homeserver; + this.transactionScopeAccessTokenHash = createHash("sha256").update(accessToken).digest("hex"); + this.transactionScopeDeviceId = opts.deviceId?.trim() || null; this.httpClient = new MatrixAuthedHttpClient({ homeserver, accessToken, @@ -146,6 +198,10 @@ export abstract class MatrixClientBase { const cryptoCallbacks = this.encryptionEnabled ? this.recoveryKeyStore.buildCryptoCallbacks() : undefined; + const guardedFetch = createMatrixGuardedFetch({ + ssrfPolicy: opts.ssrfPolicy, + dispatcherPolicy: opts.dispatcherPolicy, + }); this.client = createMatrixJsClient({ baseUrl: homeserver, accessToken, @@ -153,10 +209,13 @@ export abstract class MatrixClientBase { deviceId: opts.deviceId, logger: createMatrixJsSdkClientLogger("MatrixClient"), localTimeoutMs: this.localTimeoutMs, - fetchFn: createMatrixGuardedFetch({ - ssrfPolicy: opts.ssrfPolicy, - dispatcherPolicy: opts.dispatcherPolicy, - }), + fetchFn: (async (resource: RequestInfo | URL, init?: RequestInit) => { + const dispatch = resolveMessageWireDispatch(resource, init); + if (dispatch) { + await this.messageWireDispatchGuards.get(dispatch.transactionId)?.(dispatch); + } + return await guardedFetch(resource, init); + }) as typeof fetch, store: this.syncStore, cryptoCallbacks: cryptoCallbacks as never, verificationMethods: [ @@ -168,6 +227,25 @@ export abstract class MatrixClientBase { }); } + protected async withMessageWireDispatchGuard(params: { + transactionId?: string; + guard?: MatrixMessageWireDispatchGuard; + run: () => Promise; + }): Promise { + if (!params.transactionId || !params.guard) { + return await params.run(); + } + if (this.messageWireDispatchGuards.has(params.transactionId)) { + throw new Error(`Matrix transaction ${params.transactionId} already has a dispatch guard`); + } + this.messageWireDispatchGuards.set(params.transactionId, params.guard); + try { + return await params.run(); + } finally { + this.messageWireDispatchGuards.delete(params.transactionId); + } + } + on( eventName: TEvent, listener: (...args: MatrixClientEventMap[TEvent]) => void, diff --git a/extensions/matrix/src/matrix/sdk/client-core.ts b/extensions/matrix/src/matrix/sdk/client-core.ts index d63adf6362ea..67aba693e0bf 100644 --- a/extensions/matrix/src/matrix/sdk/client-core.ts +++ b/extensions/matrix/src/matrix/sdk/client-core.ts @@ -1,7 +1,9 @@ +import { createHash } from "node:crypto"; import { MatrixEventEvent, Preset, type MatrixEvent } from "matrix-js-sdk/lib/matrix.js"; +import { EventStatus } from "matrix-js-sdk/lib/models/event-status.js"; import type { Direction } from "matrix-js-sdk/lib/models/event-timeline.js"; import { formatMatrixErrorReason } from "../errors.js"; -import { MatrixClientBase } from "./client-base.js"; +import { MatrixClientBase, type MatrixMessageWireDispatch } from "./client-base.js"; import { matrixEventToRaw, parseMxc } from "./event-helpers.js"; import { noop } from "./logger.js"; import type { HttpMethod, QueryParams } from "./transport.js"; @@ -49,6 +51,57 @@ export abstract class MatrixClientCore extends MatrixClientBase { return Array.isArray(joined.joined_rooms) ? joined.joined_rooms : []; } + async getTransactionScopeId(): Promise { + if (this.transactionScopeId) { + return this.transactionScopeId; + } + const active = + this.transactionScopePromise ?? + (async () => { + const configuredUserId = this.client.getUserId()?.trim() || this.selfUserId; + const configuredDeviceId = + this.transactionScopeDeviceId || this.client.getDeviceId()?.trim() || null; + const whoami = (await this.doRequest("GET", "/_matrix/client/v3/account/whoami")) as { + user_id?: string; + device_id?: string; + }; + const userId = whoami.user_id?.trim() || null; + const deviceId = whoami.device_id?.trim() || null; + if (!userId) { + throw new Error("Matrix whoami did not return user_id"); + } + if (configuredUserId && configuredUserId !== userId) { + throw new Error("Matrix access token user does not match the configured userId"); + } + if (configuredDeviceId && deviceId && configuredDeviceId !== deviceId) { + throw new Error("Matrix access token device does not match the configured deviceId"); + } + this.selfUserId = userId; + this.transactionScopeDeviceId = deviceId; + // Include both device and token identities. This deliberately fails closed + // across credential rotation even where a homeserver could reuse a device txn scope. + return createHash("sha256") + .update(this.transactionScopeHomeserver) + .update("\0") + .update(userId) + .update("\0") + .update(deviceId ?? "") + .update("\0") + .update(this.transactionScopeAccessTokenHash) + .digest("hex"); + })(); + this.transactionScopePromise = active; + try { + const resolved = await active; + this.transactionScopeId = resolved; + return resolved; + } finally { + if (this.transactionScopePromise === active) { + this.transactionScopePromise = null; + } + } + } + async getJoinedRoomMembers(roomId: string): Promise { const members = await this.client.getJoinedRoomMembers(roomId); const joined = members?.joined; @@ -124,13 +177,55 @@ export abstract class MatrixClientCore extends MatrixClientBase { return result.room_id; } - async sendMessage(roomId: string, content: MessageEventContent): Promise { + async sendMessage( + roomId: string, + content: MessageEventContent, + transactionId?: string, + beforeWireDispatch?: (dispatch: MatrixMessageWireDispatch) => Promise, + ): Promise { return await this.runSerializedRoomSend(roomId, async () => { - const sent = await this.client.sendMessage(roomId, content as never); - return sent.event_id; + return await this.withMessageWireDispatchGuard({ + transactionId, + guard: beforeWireDispatch, + run: async () => { + if (transactionId) { + const room = this.client.getRoom(roomId); + const existing = room?.getEventForTxnId?.(transactionId); + if (existing) { + const existingId = existing.getId(); + if ( + existing.status === EventStatus.SENT && + existingId && + !existingId.startsWith("~") + ) { + return existingId; + } + if (existing.status === EventStatus.NOT_SENT && room) { + const resent = await this.client.resendEvent(existing, room); + return resent.event_id; + } + throw new Error( + `Matrix transaction ${transactionId} is already active with status ${existing.status ?? "unknown"}`, + ); + } + } + const sent = await this.client.sendMessage(roomId, content as never, transactionId); + return sent.event_id; + }, + }); }); } + async getMessageWireEventType(roomId: string): Promise<"m.room.message" | "m.room.encrypted"> { + if (this.client.getRoom(roomId)?.hasEncryptionStateEvent() === true) { + return "m.room.encrypted"; + } + const crypto = this.client.getCrypto(); + return crypto && (await crypto.isEncryptionEnabledInRoom(roomId)) + ? "m.room.encrypted" + : "m.room.message"; + } + async sendEvent( roomId: string, eventType: string, diff --git a/extensions/matrix/src/matrix/send.test.ts b/extensions/matrix/src/matrix/send.test.ts index 368c1336ea8f..11a867478ec4 100644 --- a/extensions/matrix/src/matrix/send.test.ts +++ b/extensions/matrix/src/matrix/send.test.ts @@ -1,8 +1,17 @@ // Matrix tests cover send plugin behavior. -import { beforeEach, describe, expect, it, vi } from "vitest"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { + resetPluginBlobStoreForTests, + resetPluginStateStoreForTests, +} from "openclaw/plugin-sdk/plugin-state-test-runtime"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import type { PluginRuntime } from "../../runtime-api.js"; import { setMatrixRuntime } from "../runtime.js"; +import { installMatrixTestRuntime } from "../test-runtime.js"; import { voteMatrixPoll } from "./actions/polls.js"; +import { loadMatrixDeliveryPlan, resolveMatrixDurableDeliveryIdentity } from "./delivery-plan.js"; import { markdownToMatrixBody, markdownToMatrixHtml } from "./format.js"; import { chunkMatrixText, @@ -115,6 +124,8 @@ const makeClient = () => { getEvent, getJoinedRoomMembers, uploadContent, + getTransactionScopeId: vi.fn().mockResolvedValue("scope-1"), + getMessageWireEventType: vi.fn().mockResolvedValue("m.room.message"), getUserId: vi.fn().mockResolvedValue("@bot:example.org"), prepareForOneOff: vi.fn(async () => undefined), start: vi.fn(async () => undefined), @@ -411,6 +422,87 @@ describe("Matrix formatted chunk boundaries", () => { }); }); +describe("sendMessageMatrix durable delivery", () => { + let stateDir = ""; + + beforeEach(() => { + resetMatrixSendRuntimeMocks(); + stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-matrix-send-plan-")); + installMatrixTestRuntime({ + stateDir, + cfg: {}, + channel: runtimeStub.channel, + }); + }); + + afterEach(() => { + resetPluginBlobStoreForTests({ closeDatabase: false }); + resetPluginStateStoreForTests(); + fs.rmSync(stateDir, { recursive: true, force: true }); + }); + + it("persists the complete event plan before the first provider dispatch", async () => { + const { client, sendMessage } = makeClient(); + const deliveryIdentity = resolveMatrixDurableDeliveryIdentity({ + queueId: "queue-1", + partIndex: 0, + partCount: 1, + }); + if (!deliveryIdentity) { + throw new Error("expected durable Matrix identity"); + } + const dispatch = vi.fn(async () => { + await expect( + loadMatrixDeliveryPlan({ + identity: deliveryIdentity, + accountId: "default", + roomId: "!room:example", + transactionScopeId: "scope-1", + wireEventType: "m.room.message", + }), + ).resolves.not.toBeNull(); + }); + sendMessage.mockImplementation( + async ( + roomId: string, + _content: unknown, + transactionId?: string, + beforeWireDispatch?: (dispatch: { + roomId: string; + eventType: "m.room.message"; + transactionId: string; + requestPath: string; + }) => Promise, + ) => { + if (!transactionId || !beforeWireDispatch) { + throw new Error("expected durable Matrix dispatch context"); + } + await beforeWireDispatch({ + roomId, + eventType: "m.room.message", + transactionId, + requestPath: `/_matrix/client/v3/rooms/${encodeURIComponent(roomId)}/send/m.room.message/${transactionId}`, + }); + return "$event-1"; + }, + ); + + const result = await sendMessageMatrix("room:!room:example", "durable", { + client, + cfg: {} as never, + accountId: "default", + deliveryQueueId: "queue-1", + deliveryPartIndex: 0, + deliveryPartCount: 1, + onPlatformSendDispatch: dispatch, + }); + + expect(result.messageId).toBe("$event-1"); + expect(dispatch).toHaveBeenCalledOnce(); + expect(sendMessage.mock.calls[0]?.[2]).toMatch(/^oc_/); + }); +}); + describe("sendMessageMatrix media", () => { beforeEach(() => { resetMatrixSendRuntimeMocks(); diff --git a/extensions/matrix/src/matrix/send.ts b/extensions/matrix/src/matrix/send.ts index b088d230aa52..482a910938aa 100644 --- a/extensions/matrix/src/matrix/send.ts +++ b/extensions/matrix/src/matrix/send.ts @@ -6,6 +6,13 @@ import { import { requireRuntimeConfig } from "openclaw/plugin-sdk/plugin-config-runtime"; import type { PollInput } from "../runtime-api.js"; import type { CoreConfig } from "../types.js"; +import { + createMatrixPlannedEvents, + loadMatrixDeliveryPlan, + persistMatrixDeliveryPlan, + resolveMatrixDurableDeliveryIdentity, + type MatrixPreparedEvent, +} from "./delivery-plan.js"; import { loadOutboundMediaFromUrl } from "./outbound-media-runtime.js"; import { buildPollStartContent, M_POLL_START } from "./poll-types.js"; import { buildMatrixReactionContent } from "./reaction-common.js"; @@ -174,6 +181,11 @@ export async function sendMessageMatrix( if (!trimmedMessage && !opts.mediaUrl) { throw new Error("Matrix send requires text or media"); } + const durableIdentity = resolveMatrixDurableDeliveryIdentity({ + queueId: opts.deliveryQueueId, + partIndex: opts.deliveryPartIndex, + partCount: opts.deliveryPartCount, + }); return await withResolvedMatrixSendClient( { client: opts.client, @@ -184,145 +196,188 @@ export async function sendMessageMatrix( async (client) => { const roomId = await resolveMatrixRoomId(client, to); const cfg = requireRuntimeConfig(opts.cfg, "Matrix send") as CoreConfig; - const { chunks, tableMode } = chunkMatrixText(trimmedMessage, { - cfg, - accountId: opts.accountId, - }); const threadId = normalizeThreadId(opts.threadId); - const relation = threadId - ? buildThreadRelation(threadId, opts.replyToId) - : buildReplyRelation(opts.replyToId); - let pendingExtraContent = opts.extraContent; - const sendContent = async (content: MatrixOutboundContent, kind: MessageReceiptPartKind) => { - const contentWithExtra = withMatrixExtraContentFields(content, pendingExtraContent); - pendingExtraContent = undefined; - const eventId = await client.sendMessage(roomId, contentWithExtra); - const visibleContent = contentWithExtra.body ?? ""; - if (eventId) { - acceptedContents.push(visibleContent); - await opts.onDeliveryResult?.({ - messageId: eventId, + const transactionScopeId = durableIdentity ? await client.getTransactionScopeId() : undefined; + const wireEventType = durableIdentity + ? await client.getMessageWireEventType(roomId) + : undefined; + const storedPlan = durableIdentity + ? await loadMatrixDeliveryPlan({ + identity: durableIdentity, + accountId: opts.accountId, roomId, - primaryMessageId: eventId, - receipt: createMatrixSendReceipt({ - roomId, - platformMessageIds: [eventId], - kind, - replyToId: opts.replyToId, - threadId, - }), - content: visibleContent, + transactionScopeId: transactionScopeId!, + wireEventType: wireEventType!, + }) + : null; + let plannedEvents: MatrixPreparedEvent[] | undefined = storedPlan?.events; + if (!plannedEvents) { + const { chunks, tableMode } = chunkMatrixText(trimmedMessage, { + cfg, + accountId: opts.accountId, + }); + const relation = threadId + ? buildThreadRelation(threadId, opts.replyToId) + : buildReplyRelation(opts.replyToId); + let pendingExtraContent = opts.extraContent; + const events: Omit[] = []; + const prepareContent = ( + content: MatrixOutboundContent, + receiptKind: MessageReceiptPartKind, + ) => { + events.push({ + content: withMatrixExtraContentFields(content, pendingExtraContent), + receiptKind, }); - } - return eventId; - }; + pendingExtraContent = undefined; + }; - const platformMessageIds: string[] = []; - const acceptedContents: string[] = []; - let lastMessageId = ""; - let receiptKind: MessageReceiptPartKind = "text"; - if (opts.mediaUrl) { - const maxBytes = resolveMediaMaxBytes(opts.accountId, cfg); - const media = await loadOutboundMediaFromUrl(opts.mediaUrl, { - maxBytes, - mediaAccess: opts.mediaAccess, - mediaLocalRoots: opts.mediaLocalRoots, - mediaReadFile: opts.mediaReadFile, - }); - const uploaded = await uploadMediaMaybeEncrypted(client, roomId, media.buffer, { - contentType: media.contentType, - filename: media.fileName, - }); - const durationMs = await resolveMediaDurationMs({ - buffer: media.buffer, - contentType: media.contentType, - fileName: media.fileName, - kind: media.kind === "sticker" ? "unknown" : (media.kind ?? "unknown"), - }); - const baseMsgType = resolveMatrixMsgType(media.contentType, media.fileName); - const { useVoice } = resolveMatrixVoiceDecision({ - wantsVoice: opts.audioAsVoice === true, - contentType: media.contentType, - fileName: media.fileName, - }); - const msgtype = useVoice ? MsgType.Audio : baseMsgType; - receiptKind = useVoice ? "voice" : "media"; - const isImage = msgtype === MsgType.Image; - const imageInfo = isImage - ? await prepareImageInfo({ - buffer: media.buffer, - client, - encrypted: Boolean(uploaded.file), - }) - : undefined; - const [firstChunk, ...rest] = chunks; - const captionMarkdown = useVoice ? "" : (firstChunk ?? ""); - const body = useVoice ? "Voice message" : captionMarkdown || media.fileName || "(file)"; - const content = buildMediaContent({ - msgtype, - body, - url: uploaded.url, - file: uploaded.file, - filename: media.fileName, - mimetype: media.contentType, - size: media.buffer.byteLength, - durationMs, - relation, - isVoice: useVoice, - imageInfo, - }); - await enrichMatrixFormattedContent({ - client, - content, - markdown: captionMarkdown, - tableMode, - }); - const eventId = await sendContent(content, receiptKind); - lastMessageId = eventId ?? lastMessageId; - if (eventId) { - platformMessageIds.push(eventId); - } - const textChunks = useVoice ? chunks : rest; - // Voice messages use a generic media body ("Voice message"), so keep any - // transcript follow-up attached to the same reply/thread context. - const followupRelation = useVoice || threadId ? relation : undefined; - for (const chunk of textChunks) { - const text = chunk; - if (!text.trim()) { - continue; - } - const followup = buildTextContent(text, followupRelation); - await enrichMatrixFormattedContent({ - client, - content: followup, - markdown: text, - tableMode, + if (opts.mediaUrl) { + const maxBytes = resolveMediaMaxBytes(opts.accountId, cfg); + const media = await loadOutboundMediaFromUrl(opts.mediaUrl, { + maxBytes, + mediaAccess: opts.mediaAccess, + mediaLocalRoots: opts.mediaLocalRoots, + mediaReadFile: opts.mediaReadFile, + }); + const uploaded = await uploadMediaMaybeEncrypted(client, roomId, media.buffer, { + contentType: media.contentType, + filename: media.fileName, + }); + const durationMs = await resolveMediaDurationMs({ + buffer: media.buffer, + contentType: media.contentType, + fileName: media.fileName, + kind: media.kind === "sticker" ? "unknown" : (media.kind ?? "unknown"), + }); + const baseMsgType = resolveMatrixMsgType(media.contentType, media.fileName); + const { useVoice } = resolveMatrixVoiceDecision({ + wantsVoice: opts.audioAsVoice === true, + contentType: media.contentType, + fileName: media.fileName, + }); + const msgtype = useVoice ? MsgType.Audio : baseMsgType; + const receiptKind: MessageReceiptPartKind = useVoice ? "voice" : "media"; + const imageInfo = + msgtype === MsgType.Image + ? await prepareImageInfo({ + buffer: media.buffer, + client, + encrypted: Boolean(uploaded.file), + }) + : undefined; + const [firstChunk, ...rest] = chunks; + const captionMarkdown = useVoice ? "" : (firstChunk ?? ""); + const content = buildMediaContent({ + msgtype, + body: useVoice ? "Voice message" : captionMarkdown || media.fileName || "(file)", + url: uploaded.url, + file: uploaded.file, + filename: media.fileName, + mimetype: media.contentType, + size: media.buffer.byteLength, + durationMs, + relation, + isVoice: useVoice, + imageInfo, }); - const followupEventId = await sendContent(followup, "text"); - lastMessageId = followupEventId ?? lastMessageId; - if (followupEventId) { - platformMessageIds.push(followupEventId); - } - } - } else { - for (const chunk of chunks.length ? chunks : [""]) { - const text = chunk; - if (!text.trim()) { - continue; - } - const content = buildTextContent(text, relation); await enrichMatrixFormattedContent({ client, content, - markdown: text, + markdown: captionMarkdown, tableMode, }); - const eventId = await sendContent(content, "text"); - lastMessageId = eventId ?? lastMessageId; - if (eventId) { - platformMessageIds.push(eventId); + prepareContent(content, receiptKind); + const textChunks = useVoice ? chunks : rest; + const followupRelation = useVoice || threadId ? relation : undefined; + for (const chunk of textChunks) { + if (!chunk.trim()) { + continue; + } + const followup = buildTextContent(chunk, followupRelation); + await enrichMatrixFormattedContent({ + client, + content: followup, + markdown: chunk, + tableMode, + }); + prepareContent(followup, "text"); + } + } else { + for (const chunk of chunks.length ? chunks : [""]) { + if (!chunk.trim()) { + continue; + } + const content = buildTextContent(chunk, relation); + await enrichMatrixFormattedContent({ + client, + content, + markdown: chunk, + tableMode, + }); + prepareContent(content, "text"); } } + plannedEvents = durableIdentity + ? createMatrixPlannedEvents({ identity: durableIdentity, events }) + : events.map((event) => ({ + content: event.content, + receiptKind: event.receiptKind, + transactionId: "", + })); + } + + let platformDispatchStarted = false; + if (!durableIdentity) { + await opts.onPlatformSendDispatch?.(); + platformDispatchStarted = true; + } + const platformMessageIds: string[] = []; + const acceptedContents: string[] = []; + let lastMessageId = ""; + for (const planned of plannedEvents) { + const eventId = await client.sendMessage( + roomId, + planned.content, + planned.transactionId || undefined, + durableIdentity + ? async (dispatch) => { + await persistMatrixDeliveryPlan({ + identity: durableIdentity, + accountId: opts.accountId, + roomId, + transactionScopeId: transactionScopeId!, + wireEventType: dispatch.eventType, + events: plannedEvents, + dispatch, + }); + if (!platformDispatchStarted) { + await opts.onPlatformSendDispatch?.(); + platformDispatchStarted = true; + } + } + : undefined, + ); + lastMessageId = eventId || lastMessageId; + if (!eventId) { + continue; + } + platformMessageIds.push(eventId); + const visibleContent = planned.content.body ?? ""; + acceptedContents.push(visibleContent); + await opts.onDeliveryResult?.({ + messageId: eventId, + roomId, + primaryMessageId: eventId, + receipt: createMatrixSendReceipt({ + roomId, + platformMessageIds: [eventId], + kind: planned.receiptKind, + replyToId: opts.replyToId, + threadId, + }), + content: visibleContent, + }); } return { @@ -332,7 +387,7 @@ export async function sendMessageMatrix( receipt: createMatrixSendReceipt({ roomId, platformMessageIds, - kind: receiptKind, + kind: plannedEvents[0]?.receiptKind ?? "text", replyToId: opts.replyToId, threadId, }), diff --git a/extensions/matrix/src/matrix/send/types.ts b/extensions/matrix/src/matrix/send/types.ts index 3ceaab510ba9..bf0bc5441772 100644 --- a/extensions/matrix/src/matrix/send/types.ts +++ b/extensions/matrix/src/matrix/send/types.ts @@ -100,6 +100,14 @@ export type MatrixSendOpts = { replyToId?: string; threadId?: string | number | null; timeoutMs?: number; + /** Opaque durable queue id used to derive Matrix transaction ids. */ + deliveryQueueId?: string; + /** Stable provider-send index within one durable payload. */ + deliveryPartIndex?: number; + /** Exact provider-send count within one durable payload. */ + deliveryPartCount?: number; + /** Marks recipient-visible timeline dispatch after the recovery plan is durable. */ + onPlatformSendDispatch?: () => Promise; /** Additional Matrix event content fields to merge into the first sent event. */ extraContent?: MatrixExtraContentFields; /** Send audio as voice message instead of audio file. Defaults to false. */ diff --git a/extensions/matrix/src/outbound.ts b/extensions/matrix/src/outbound.ts index 2171cb78a87a..a1efb2a4dd6b 100644 --- a/extensions/matrix/src/outbound.ts +++ b/extensions/matrix/src/outbound.ts @@ -210,6 +210,10 @@ export const matrixOutbound: ChannelOutboundAdapter = { threadId, accountId, audioAsVoice, + deliveryQueueId, + deliveryPartIndex, + deliveryPartCount, + onPlatformSendDispatch, onDeliveryResult, }) => { const send = @@ -222,6 +226,10 @@ export const matrixOutbound: ChannelOutboundAdapter = { threadId: resolvedThreadId, accountId: accountId ?? undefined, audioAsVoice, + deliveryQueueId, + deliveryPartIndex, + ...(deliveryQueueId !== undefined ? { deliveryPartCount } : {}), + onPlatformSendDispatch, onDeliveryResult: resolveMatrixDeliveryProgress(onDeliveryResult), }); return { @@ -242,6 +250,10 @@ export const matrixOutbound: ChannelOutboundAdapter = { threadId, accountId, audioAsVoice, + deliveryQueueId, + deliveryPartIndex, + deliveryPartCount, + onPlatformSendDispatch, onDeliveryResult, }) => { const send = @@ -257,6 +269,10 @@ export const matrixOutbound: ChannelOutboundAdapter = { threadId: resolvedThreadId, accountId: accountId ?? undefined, audioAsVoice, + deliveryQueueId, + deliveryPartIndex, + ...(deliveryQueueId !== undefined ? { deliveryPartCount } : {}), + onPlatformSendDispatch, onDeliveryResult: resolveMatrixDeliveryProgress(onDeliveryResult), }); return { diff --git a/extensions/matrix/src/test-runtime.ts b/extensions/matrix/src/test-runtime.ts index e08687f13702..8cd058dd11a0 100644 --- a/extensions/matrix/src/test-runtime.ts +++ b/extensions/matrix/src/test-runtime.ts @@ -3,8 +3,12 @@ import { implicitMentionKindWhen, resolveInboundMentionDecision, } from "openclaw/plugin-sdk/channel-mention-gating"; -import type { OpenKeyedStoreOptions } from "openclaw/plugin-sdk/plugin-state-runtime"; +import type { + OpenBlobStoreOptions, + OpenKeyedStoreOptions, +} from "openclaw/plugin-sdk/plugin-state-runtime"; import { + createPluginBlobStoreForTests, createPluginStateKeyedStoreForTests, createPluginStateSyncKeyedStoreForTests, } from "openclaw/plugin-sdk/plugin-state-test-runtime"; @@ -25,7 +29,7 @@ type MatrixRuntimeStub = { logging?: PluginRuntime["logging"]; state: Pick< NonNullable, - "openKeyedStore" | "openSyncKeyedStore" | "resolveStateDir" + "openBlobStore" | "openKeyedStore" | "openSyncKeyedStore" | "resolveStateDir" >; }; @@ -89,6 +93,11 @@ export function installMatrixTestRuntime(options: MatrixTestRuntimeOptions = {}) ...(logging ? { logging } : {}), state: { resolveStateDir: defaultStateDirResolver, + openBlobStore: ((storeOptions: OpenBlobStoreOptions) => + createPluginBlobStoreForTests("matrix", storeOptions, { + ...process.env, + OPENCLAW_STATE_DIR: defaultStateDirResolver(process.env, osHomedirForTest), + })) as PluginRuntime["state"]["openBlobStore"], openKeyedStore: ((storeOptions: OpenKeyedStoreOptions) => createPluginStateKeyedStoreForTests("matrix", { ...storeOptions, diff --git a/src/channels/message/types.ts b/src/channels/message/types.ts index 75ad7f8ea4f2..969be50bc114 100644 --- a/src/channels/message/types.ts +++ b/src/channels/message/types.ts @@ -183,6 +183,8 @@ export type ChannelMessageSendTextContext = { deliveryQueueId?: string; /** @internal Stable platform-send index within one durable payload. */ deliveryPartIndex?: number; + /** @internal Exact platform-send count within one durable payload. */ + deliveryPartCount?: number; /** @internal Channel-valid id reserved before a correlated conversation turn is sent. */ preparedMessageId?: string; /** @internal Refresh durable timing before recipient-visible or finalizing platform I/O. */ @@ -355,6 +357,8 @@ type ChannelMessageSendAdapter< /** Durable final-delivery extension for queue reconciliation and capability declaration. */ export type ChannelMessageDurableFinalAdapter = { capabilities?: DurableFinalDeliveryRequirementMap; + /** Opt into provider reconciliation for ordinary single-payload queued sends. */ + automaticUnknownSendReconciliation?: boolean; /** * Synchronous provider admission before a durable intent is created or replayed. * Providers must not perform I/O from this hook. @@ -370,6 +374,8 @@ export type ChannelMessageDurableFinalAdapter = { | Promise | ChannelMessageUnknownSendReconciliationResult | null; + /** Cleanup after core authoritatively retires an ambiguous send as failed. */ + afterUnknownSendTerminal?: (ctx: ChannelMessageUnknownSendContext) => Promise | void; }; /** Live-message feature key declared by adapters that support preview or streaming behavior. */ diff --git a/src/channels/plugins/outbound.types.ts b/src/channels/plugins/outbound.types.ts index 0a7e9e71a012..8a34b43ecd82 100644 --- a/src/channels/plugins/outbound.types.ts +++ b/src/channels/plugins/outbound.types.ts @@ -44,6 +44,8 @@ export type ChannelOutboundContext = { deliveryQueueId?: string; /** @internal Stable platform-send index within one durable payload. */ deliveryPartIndex?: number; + /** @internal Exact platform-send count within one durable payload. */ + deliveryPartCount?: number; /** @internal Channel-valid id reserved before a correlated conversation turn is sent. */ preparedMessageId?: string; /** @internal Refresh durable timing before recipient-visible or finalizing platform I/O. */ diff --git a/src/cli/send-runtime/channel-outbound-send.test.ts b/src/cli/send-runtime/channel-outbound-send.test.ts index 7ca6a95d7bfa..b73f8dacc345 100644 --- a/src/cli/send-runtime/channel-outbound-send.test.ts +++ b/src/cli/send-runtime/channel-outbound-send.test.ts @@ -83,6 +83,8 @@ describe("createChannelOutboundRuntimeSend", () => { cfg: {}, accountId: "default", deliveryQueueId: "queue-1", + deliveryPartIndex: 3, + deliveryPartCount: 4, onPlatformSendDispatch, }); @@ -92,6 +94,8 @@ describe("createChannelOutboundRuntimeSend", () => { expect(params.text).toBe("hello"); expect(params.accountId).toBe("default"); expect(params.deliveryQueueId).toBe("queue-1"); + expect(params.deliveryPartIndex).toBe(3); + expect(params.deliveryPartCount).toBe(4); expect(params.onPlatformSendDispatch).toBe(onPlatformSendDispatch); }); diff --git a/src/cli/send-runtime/channel-outbound-send.ts b/src/cli/send-runtime/channel-outbound-send.ts index f41f16f8d13e..96c9400ccff0 100644 --- a/src/cli/send-runtime/channel-outbound-send.ts +++ b/src/cli/send-runtime/channel-outbound-send.ts @@ -27,6 +27,10 @@ type RuntimeSendOpts = { gatewayClientScopes?: readonly string[]; /** @internal Opaque durable intent id for provider-side reconciliation. */ deliveryQueueId?: string; + /** @internal Stable provider-send index within one payload. */ + deliveryPartIndex?: number; + /** @internal Exact provider-send count for one payload. */ + deliveryPartCount?: number; /** @internal Refresh durable timing before recipient-visible or finalizing platform I/O. */ onPlatformSendDispatch?: () => Promise; textMode?: "markdown" | "html"; @@ -70,6 +74,8 @@ export function createChannelOutboundRuntimeSend(params: { gifPlayback: opts.gifPlayback, gatewayClientScopes: opts.gatewayClientScopes, deliveryQueueId: opts.deliveryQueueId, + deliveryPartIndex: opts.deliveryPartIndex, + deliveryPartCount: opts.deliveryPartCount, onPlatformSendDispatch: opts.onPlatformSendDispatch, }); const hasMedia = Boolean(opts.mediaUrl); diff --git a/src/infra/heartbeat-runner.ack-token-heartbeat-acks.test.ts b/src/infra/heartbeat-runner.ack-token-heartbeat-acks.test.ts index 3b790580913a..52976f68e3f7 100644 --- a/src/infra/heartbeat-runner.ack-token-heartbeat-acks.test.ts +++ b/src/infra/heartbeat-runner.ack-token-heartbeat-acks.test.ts @@ -116,6 +116,8 @@ describe("runHeartbeatOnce ack handling", () => { cfg: params.cfg, accountId: undefined, audioAsVoice: undefined, + conversationReadOrigin: undefined, + deliveryPartCount: 1, deliveryPartIndex: 0, deliveryQueueId: undefined, forceDocument: undefined, @@ -129,6 +131,7 @@ describe("runHeartbeatOnce ack handling", () => { mediaReadFile: undefined, onDeliveryResult: expect.any(Function), onPlatformSendDispatch: expect.any(Function), + preparedMessageId: undefined, replyToIdSource: undefined, replyToMode: undefined, silent: undefined, diff --git a/src/infra/outbound/deliver-channel.ts b/src/infra/outbound/deliver-channel.ts index 44de009c09b9..02c433a1a661 100644 --- a/src/infra/outbound/deliver-channel.ts +++ b/src/infra/outbound/deliver-channel.ts @@ -179,7 +179,11 @@ export async function resolveOutboundDurableFinalDeliverySupport(params: { } } - return { ok: true }; + return { + ok: true, + automaticUnknownSendReconciliation: + messageDurableFinal?.automaticUnknownSendReconciliation === true, + }; } function createPluginHandler( @@ -235,6 +239,7 @@ function createPluginHandler( threadId: overrides && "threadId" in overrides ? overrides.threadId : baseCtx.threadId, audioAsVoice: overrides?.audioAsVoice, deliveryPartIndex: overrides?.deliveryPartIndex, + deliveryPartCount: overrides?.deliveryPartCount, preparedMessageId: overrides?.deliveryPartIndex === undefined || overrides.deliveryPartIndex === 0 ? baseCtx.preparedMessageId diff --git a/src/infra/outbound/deliver-contracts.ts b/src/infra/outbound/deliver-contracts.ts index 97194d382e9b..c3795da3a17d 100644 --- a/src/infra/outbound/deliver-contracts.ts +++ b/src/infra/outbound/deliver-contracts.ts @@ -46,7 +46,7 @@ export type DurableFinalDeliveryRequirements = Partial< >; export type OutboundDurableDeliverySupport = - | { ok: true } + | { ok: true; automaticUnknownSendReconciliation: boolean } | { ok: false; reason: "missing_outbound_handler" | "capability_mismatch"; diff --git a/src/infra/outbound/deliver-queue.ts b/src/infra/outbound/deliver-queue.ts index e3eae24694e6..2444b173e0e7 100644 --- a/src/infra/outbound/deliver-queue.ts +++ b/src/infra/outbound/deliver-queue.ts @@ -166,15 +166,8 @@ async function runOutboundDeliveryWithQueue( existingStableDelivery?.renderedBatchPlan ?? (params.preparedBatch ? params.renderedBatchPlan : undefined) ?? createRenderedMessageBatchPlan(preparedPayloads); - const deliveryParams: DeliverOutboundPayloadsParams = { - ...params, - payloads: preparedPayloads, - preparedBatch, - // Recovery must preserve the provider-facing plan captured before local - // media was rewritten to spool paths; reconciliation uses that same plan. - renderedBatchPlan: preparedRenderedBatchPlan, - }; - if (params.requireUnknownSendReconciliation === true) { + let unknownSendReconciliationEnabled = params.requireUnknownSendReconciliation === true; + if (params.requireUnknownSendReconciliation !== false && preparedPayloads.length === 1) { const requirements = deriveDurableFinalDeliveryRequirementsForBatch({ payloads: preparedPayloads, replyToId: params.replyToId, @@ -188,13 +181,26 @@ async function runOutboundDeliveryWithQueue( channel, requirements, }); - if (!support.ok) { + if (params.requireUnknownSendReconciliation === true && !support.ok) { emitPreQueueFailure(); throw new Error( `Required durable message send is unsupported for ${channel}: prepared payload capability mismatch${support.capability ? ` (${support.capability})` : ""}`, ); } + unknownSendReconciliationEnabled = + support.ok && + (params.requireUnknownSendReconciliation === true || + support.automaticUnknownSendReconciliation); } + const deliveryParams: DeliverOutboundPayloadsParams = { + ...params, + payloads: preparedPayloads, + preparedBatch, + // Recovery must preserve the provider-facing plan captured before local + // media was rewritten to spool paths; reconciliation uses that same plan. + renderedBatchPlan: preparedRenderedBatchPlan, + ...(unknownSendReconciliationEnabled ? { requireUnknownSendReconciliation: true } : {}), + }; // Invocation authority is not queued; recovery must re-enter delegated after restart. // Write-ahead delivery queue: persist before sending, remove after success. diff --git a/src/infra/outbound/deliver.test.ts b/src/infra/outbound/deliver.test.ts index 8e8882c287d8..dd6dc41b45cf 100644 --- a/src/infra/outbound/deliver.test.ts +++ b/src/infra/outbound/deliver.test.ts @@ -604,7 +604,7 @@ describe("deliverOutboundPayloads", () => { silent: true, }, }), - ).resolves.toEqual({ ok: true }); + ).resolves.toEqual({ ok: true, automaticUnknownSendReconciliation: false }); }); it("requires a real reconciler for required unknown-send recovery support", async () => { @@ -687,7 +687,7 @@ describe("deliverOutboundPayloads", () => { reconcileUnknownSend: true, }, }), - ).resolves.toEqual({ ok: true }); + ).resolves.toEqual({ ok: true, automaticUnknownSendReconciliation: false }); await expect( resolveOutboundDurableFinalDeliverySupport({ @@ -729,7 +729,7 @@ describe("deliverOutboundPayloads", () => { channel: "matrix", requirements: { text: true, reconcileUnknownSend: true }, }), - ).resolves.toEqual({ ok: true }); + ).resolves.toEqual({ ok: true, automaticUnknownSendReconciliation: false }); }); it("requires every concrete reconciliation kind for heterogeneous batches", async () => { @@ -1194,6 +1194,73 @@ describe("deliverOutboundPayloads", () => { } }); + it("automatically enables provider reconciliation for one supported prepared payload", async () => { + const messageSendText = vi.fn(async (ctx: ChannelMessageSendTextContext) => { + await ctx.onPlatformSendDispatch?.(); + return { + messageId: "message-adapter-1", + receipt: createMessageReceiptFromOutboundResults({ + results: [{ channel: "matrix", messageId: "message-adapter-1" }], + kind: "text", + }), + }; + }); + setMatrixMessageAdapter({ + id: "matrix", + durableFinal: { + automaticUnknownSendReconciliation: true, + capabilities: { text: true, reconcileUnknownSend: true }, + reconcileUnknownSendKinds: { text: true }, + reconcileUnknownSend: async () => ({ status: "not_sent" }), + }, + send: { text: messageSendText }, + }); + + await deliverMatrix({ queuePolicy: "required" }); + + expect(requireMockCallArg(queueMocks.enqueueDelivery, "enqueueDelivery")).toMatchObject({ + requireUnknownSendReconciliation: true, + }); + expect(messageSendText).toHaveBeenCalledWith( + expect.objectContaining({ + deliveryQueueId: "mock-queue-id", + deliveryPartIndex: 0, + deliveryPartCount: 1, + }), + ); + }); + + it("leaves ordinary multi-payload delivery on the existing fail-closed path", async () => { + const messageSendText = vi.fn(async (_ctx: ChannelMessageSendTextContext) => ({ + messageId: "message-adapter-1", + receipt: createMessageReceiptFromOutboundResults({ + results: [{ channel: "matrix", messageId: "message-adapter-1" }], + kind: "text", + }), + })); + setMatrixMessageAdapter({ + id: "matrix", + durableFinal: { + automaticUnknownSendReconciliation: true, + capabilities: { text: true, reconcileUnknownSend: true }, + reconcileUnknownSendKinds: { text: true }, + reconcileUnknownSend: async () => ({ status: "not_sent" }), + }, + send: { text: messageSendText }, + }); + + await deliverMatrix({ + payloads: [{ text: "first" }, { text: "second" }], + queuePolicy: "required", + }); + + expect(messageSendText).toHaveBeenCalledTimes(2); + expect(messageSendText.mock.calls.map(([ctx]) => ctx.deliveryQueueId)).toEqual([ + undefined, + undefined, + ]); + }); + it("rejects explicitly reconciled multi-payload sends before enqueue or platform I/O", async () => { const messageSendText = vi.fn(); setMatrixMessageAdapter({ diff --git a/src/infra/outbound/delivery-queue-reconciliation.ts b/src/infra/outbound/delivery-queue-reconciliation.ts index dd5170175c67..114c36a66707 100644 --- a/src/infra/outbound/delivery-queue-reconciliation.ts +++ b/src/infra/outbound/delivery-queue-reconciliation.ts @@ -1,5 +1,6 @@ import type { ReplyPayload } from "../../auto-reply/types.js"; import type { + ChannelMessageUnknownSendContext, ChannelMessageUnknownSendReconciliationResult, RenderedMessageBatchPlan, } from "../../channels/message/types.js"; @@ -24,6 +25,35 @@ type UnknownSendQueueEntry = { silent?: boolean; }; +export function buildUnknownSendContext(params: { + entry: UnknownSendQueueEntry; + payloads: readonly ReplyPayload[]; + cfg: OpenClawConfig; +}): ChannelMessageUnknownSendContext { + const { entry } = params; + return { + cfg: params.cfg, + queueId: entry.id, + channel: entry.channel, + to: entry.to, + ...(entry.accountId !== undefined ? { accountId: entry.accountId } : {}), + enqueuedAt: entry.enqueuedAt, + retryCount: entry.retryCount, + ...(entry.platformSendStartedAt !== undefined + ? { platformSendStartedAt: entry.platformSendStartedAt } + : {}), + ...(entry.effectiveReplyToId !== undefined + ? { effectiveReplyToId: entry.effectiveReplyToId } + : {}), + payloads: params.payloads, + ...(entry.renderedBatchPlan ? { renderedBatchPlan: entry.renderedBatchPlan } : {}), + ...(entry.replyToId !== undefined ? { replyToId: entry.replyToId } : {}), + ...(entry.replyToMode !== undefined ? { replyToMode: entry.replyToMode } : {}), + ...(entry.threadId !== undefined ? { threadId: entry.threadId } : {}), + ...(entry.silent !== undefined ? { silent: entry.silent } : {}), + }; +} + /** Reconciles provider state without applying or rediscovering outbound policy. */ export async function reconcileUnknownQueuedDelivery(params: { entry: UnknownSendQueueEntry; @@ -45,27 +75,7 @@ export async function reconcileUnknownQueuedDelivery(params: { } const { entry } = params; try { - return await reconcileUnknownSend({ - cfg: params.cfg, - queueId: entry.id, - channel: entry.channel, - to: entry.to, - ...(entry.accountId !== undefined ? { accountId: entry.accountId } : {}), - enqueuedAt: entry.enqueuedAt, - retryCount: entry.retryCount, - ...(entry.platformSendStartedAt !== undefined - ? { platformSendStartedAt: entry.platformSendStartedAt } - : {}), - ...(entry.effectiveReplyToId !== undefined - ? { effectiveReplyToId: entry.effectiveReplyToId } - : {}), - payloads: params.payloads, - ...(entry.renderedBatchPlan ? { renderedBatchPlan: entry.renderedBatchPlan } : {}), - ...(entry.replyToId !== undefined ? { replyToId: entry.replyToId } : {}), - ...(entry.replyToMode !== undefined ? { replyToMode: entry.replyToMode } : {}), - ...(entry.threadId !== undefined ? { threadId: entry.threadId } : {}), - ...(entry.silent !== undefined ? { silent: entry.silent } : {}), - }); + return await reconcileUnknownSend(buildUnknownSendContext(params)); } catch (error) { const message = formatErrorMessage(error); params.warn(`Delivery entry ${entry.id} unknown-send reconciliation failed: ${message}`); diff --git a/src/infra/outbound/delivery-queue-recovery.ts b/src/infra/outbound/delivery-queue-recovery.ts index b61e27cc9561..5dd2c4dcefcc 100644 --- a/src/infra/outbound/delivery-queue-recovery.ts +++ b/src/infra/outbound/delivery-queue-recovery.ts @@ -49,7 +49,10 @@ import { cancelDeliveryQueueMediaRecoveryLease, createDeliveryQueueMediaRecoveryLease, } from "./delivery-queue-media-staging.js"; -import { reconcileUnknownQueuedDelivery } from "./delivery-queue-reconciliation.js"; +import { + buildUnknownSendContext, + reconcileUnknownQueuedDelivery, +} from "./delivery-queue-reconciliation.js"; import { claimDeliveryPlatformSendAttempt, failDelivery, @@ -359,6 +362,11 @@ async function applyRecoveryDeliveryAdmission(params: { params.stateDir, ); if (result.status === "failed") { + await runUnknownSendTerminalCleanup({ + entry: params.entry, + cfg: params.cfg, + log: params.log, + }); emitRecoveredTerminalFailure(params.entry, admission.reason); emitQueuedAuditTerminals(params.entry, () => queuedDeadLetterAuditTerminals(params.entry)); params.log.warn( @@ -372,6 +380,53 @@ async function applyRecoveryDeliveryAdmission(params: { return "not_pending"; } +async function runUnknownSendTerminalCleanup(params: { + entry: QueuedDelivery; + cfg: OpenClawConfig; + log: RecoveryLogger; +}): Promise { + if (!needsUnknownSendReconciliation(params.entry)) { + return; + } + const adapter = resolveOutboundChannelMessageAdapter({ + channel: params.entry.channel, + cfg: params.cfg, + allowBootstrap: true, + }); + const cleanup = adapter?.durableFinal?.afterUnknownSendTerminal; + if (!cleanup) { + return; + } + try { + await cleanup( + buildUnknownSendContext({ + entry: params.entry, + payloads: queuedPayloads(params.entry), + cfg: params.cfg, + }), + ); + } catch (error) { + params.log.warn( + `Delivery entry ${params.entry.id} unknown-send terminal cleanup failed: ${formatErrorMessage(error)}`, + ); + } +} + +async function moveEntryToFailedAndCleanup(params: { + entry: QueuedDelivery; + cfg: OpenClawConfig; + log: RecoveryLogger; + stateDir?: string; + attemptId?: string | null; +}): Promise { + await (params.attemptId !== undefined + ? moveToFailed(params.entry.id, params.stateDir, params.attemptId) + : moveToFailed(params.entry.id, params.stateDir)); + // Cleanup follows the authoritative queue transition. Deleting provider + // evidence first could strand a still-pending ambiguous send without proof. + await runUnknownSendTerminalCleanup(params); +} + function buildReconciledSentResult( entry: QueuedDelivery, reconciliation: Extract, @@ -404,6 +459,7 @@ function buildReconciledCommitContext(params: { const base = { cfg: params.cfg, to: params.entry.to, + deliveryQueueId: params.entry.id, accountId: params.entry.accountId, replyToId: params.entry.effectiveReplyToId !== undefined @@ -483,15 +539,14 @@ async function runReconciledSentCommitHooks(params: { async function moveEntryToFailedWithLogging( entry: QueuedDelivery, + cfg: OpenClawConfig, log: RecoveryLogger, stateDir?: string, ): Promise { markDurableDeliveryFailedBestEffort(entry, log); try { const attemptId = recoveryPlatformAttemptId(entry); - await (attemptId !== undefined - ? moveToFailed(entry.id, stateDir, attemptId) - : moveToFailed(entry.id, stateDir)); + await moveEntryToFailedAndCleanup({ entry, cfg, log, stateDir, attemptId }); emitRecoveredTerminalFailure(entry, "delivery retry budget exhausted"); return true; } catch (err) { @@ -555,6 +610,7 @@ function markDurableDeliveryFailedBestEffort(entry: QueuedDelivery, log: Recover async function resolveCompletedOwnerBeforeRecovery(opts: { entry: QueuedDelivery; + cfg: OpenClawConfig; log: RecoveryLogger; stateDir?: string; onRecovered?: (entry: QueuedDelivery) => void; @@ -645,7 +701,7 @@ async function resolveCompletedOwnerBeforeRecovery(opts: { return "failed"; } if (operation.status === "unknown") { - const moved = await moveEntryToFailedWithLogging(opts.entry, opts.log, opts.stateDir); + const moved = await moveEntryToFailedWithLogging(opts.entry, opts.cfg, opts.log, opts.stateDir); return moved ? "moved-to-failed" : "failed"; } return "continue"; @@ -790,9 +846,13 @@ async function drainQueuedEntry(opts: { try { markDurableDeliveryFailedBestEffort(entry, opts.log); const attemptId = recoveryPlatformAttemptId(entry); - await (attemptId !== undefined - ? moveToFailed(entry.id, opts.stateDir, attemptId) - : moveToFailed(entry.id, opts.stateDir)); + await moveEntryToFailedAndCleanup({ + entry, + cfg: opts.cfg, + log: opts.log, + stateDir: opts.stateDir, + attemptId, + }); emitRecoveredTerminalFailure(entry, errMsg); emitQueuedAuditTerminals(entry, () => queuedUnknownAuditTerminals(entry)); return "moved-to-failed"; @@ -863,9 +923,13 @@ async function drainQueuedEntry(opts: { const errMsg = `delivery retry budget exhausted (${reservation.attemptCount}/${maxRetries})`; markDurableDeliveryFailedBestEffort(entry, opts.log); try { - await (producerClaimId - ? moveToFailed(entry.id, opts.stateDir, producerClaimId) - : moveToFailed(entry.id, opts.stateDir)); + await moveEntryToFailedAndCleanup({ + entry, + cfg: opts.cfg, + log: opts.log, + stateDir: opts.stateDir, + attemptId: producerClaimId, + }); emitRecoveredTerminalFailure(entry, errMsg); } catch (moveErr) { if (getErrnoCode(moveErr) === "ENOENT") { @@ -1089,9 +1153,13 @@ async function drainQueuedEntry(opts: { } else { markDurableDeliveryFailedBestEffort(entry, opts.log); } - await (producerClaimId - ? moveToFailed(entry.id, opts.stateDir, producerClaimId) - : moveToFailed(entry.id, opts.stateDir)); + await moveEntryToFailedAndCleanup({ + entry, + cfg: opts.cfg, + log: opts.log, + stateDir: opts.stateDir, + attemptId: producerClaimId, + }); emitRecoveredTerminalFailure(entry, errMsg, messageSentEvents); emitQueuedAuditTerminals(entry, () => failedOutboundAuditTerminals({ @@ -1184,9 +1252,13 @@ export async function drainPendingDeliveries(opts: { try { markDurableDeliveryFailedBestEffort(currentEntry, opts.log); const attemptId = recoveryPlatformAttemptId(currentEntry); - await (attemptId !== undefined - ? moveToFailed(currentEntry.id, opts.stateDir, attemptId) - : moveToFailed(currentEntry.id, opts.stateDir)); + await moveEntryToFailedAndCleanup({ + entry: currentEntry, + cfg: opts.cfg, + log: opts.log, + stateDir: opts.stateDir, + attemptId, + }); emitRecoveredTerminalFailure(currentEntry, "delivery retry budget exhausted"); } catch (err) { if (getErrnoCode(err) === "ENOENT") { @@ -1314,6 +1386,7 @@ export async function recoverPendingDeliveries(opts: { ); const movedToFailed = await moveEntryToFailedWithLogging( currentEntry, + opts.cfg, opts.log, opts.stateDir, ); diff --git a/src/infra/outbound/delivery-queue.recovery.test.ts b/src/infra/outbound/delivery-queue.recovery.test.ts index 8cfa33a34ef1..d7344e3fd6a2 100644 --- a/src/infra/outbound/delivery-queue.recovery.test.ts +++ b/src/infra/outbound/delivery-queue.recovery.test.ts @@ -1214,6 +1214,7 @@ describe("delivery-queue recovery", () => { expect(reconcileInput.retryCount).toBe(0); const afterCommitInput = mockCallArg(afterCommit) as { + deliveryQueueId?: string; kind?: string; to?: string; accountId?: string; @@ -1222,6 +1223,7 @@ describe("delivery-queue recovery", () => { silent?: boolean; result?: { messageId?: string }; }; + expect(afterCommitInput.deliveryQueueId).toBe(id); expect(afterCommitInput.kind).toBe("text"); expect(afterCommitInput.to).toBe("+1"); expect(afterCommitInput.accountId).toBe("acct-1"); @@ -1323,10 +1325,15 @@ describe("delivery-queue recovery", () => { error: "provider lookup timed out", retryable: true, }); + const afterUnknownSendTerminal = vi.fn(async (ctx: { queueId: string }) => { + expect(ctx.queueId).toBe(id); + expect(readOutboundQueueStatus(tmpDir(), id)).toBe("failed"); + }); resolveOutboundChannelMessageAdapterMock.mockReturnValue({ durableFinal: { capabilities: { reconcileUnknownSend: true }, reconcileUnknownSend, + afterUnknownSendTerminal, }, }); const deliver = vi.fn().mockResolvedValue([]); @@ -1338,6 +1345,7 @@ describe("delivery-queue recovery", () => { expect(result).toMatchObject({ failed: 1, skippedMaxRetries: 0 }); expect(await loadPendingDeliveries(tmpDir())).toHaveLength(0); expect(readOutboundQueueStatus(tmpDir(), id)).toBe("failed"); + expect(afterUnknownSendTerminal).toHaveBeenCalledOnce(); }); it("does not reconcile unknown-after-send entries unless the adapter declares the capability", async () => { diff --git a/src/infra/outbound/message-plan.test.ts b/src/infra/outbound/message-plan.test.ts index 2b5fa2d68f28..62327be0ed4a 100644 --- a/src/infra/outbound/message-plan.test.ts +++ b/src/infra/outbound/message-plan.test.ts @@ -30,6 +30,7 @@ describe("outbound message planning", () => { ["text", "ab", "reply-1"], ["text", "cd", undefined], ]); + expect(units.map((unit) => unit.overrides.deliveryPartCount)).toEqual([2, 2]); }); it("keeps explicit text replies from consuming the implicit slot", () => { @@ -85,12 +86,13 @@ describe("outbound message planning", () => { unit.mediaUrl, unit.overrides.replyToId, unit.overrides.deliveryPartIndex, + unit.overrides.deliveryPartCount, ] : [unit.kind], ), ).toEqual([ - ["media", "caption", "https://example.com/1.png", "reply-1", 0], - ["media", undefined, "https://example.com/2.png", undefined, 1], + ["media", "caption", "https://example.com/1.png", "reply-1", 0, 2], + ["media", undefined, "https://example.com/2.png", undefined, 1, 2], ]); }); @@ -107,7 +109,11 @@ describe("outbound message planning", () => { { kind: "text", text: "bold", - overrides: { formatting: { parseMode: "HTML" }, deliveryPartIndex: 0 }, + overrides: { + formatting: { parseMode: "HTML" }, + deliveryPartIndex: 0, + deliveryPartCount: 1, + }, }, ]); }); diff --git a/src/infra/outbound/message-plan.ts b/src/infra/outbound/message-plan.ts index 130b1e160458..031fa4ad7b55 100644 --- a/src/infra/outbound/message-plan.ts +++ b/src/infra/outbound/message-plan.ts @@ -18,6 +18,8 @@ export type OutboundMessageSendOverrides = ReplyToOverride & { formatting?: OutboundDeliveryFormattingOptions; /** Stable zero-based platform-send index within one durable payload. */ deliveryPartIndex?: number; + /** Exact platform-send count for this payload. */ + deliveryPartCount?: number; }; /** @@ -131,8 +133,16 @@ export function planOutboundTextMessageUnits(params: { }; }; + const withDeliveryTopology = (units: OutboundMessageUnit[]): OutboundMessageUnit[] => { + const deliveryPartCount = units.length; + return units.map((unit) => ({ + ...unit, + overrides: { ...unit.overrides, deliveryPartCount }, + })); + }; + if (!params.chunker || params.textLimit === undefined) { - return [planTextUnit(params.text, 0)]; + return withDeliveryTopology([planTextUnit(params.text, 0)]); } if (params.chunkMode === "newline") { @@ -160,15 +170,17 @@ export function planOutboundTextMessageUnits(params: { units.push(planChunkedTextUnit(chunk, units.length)); } } - return units; + return withDeliveryTopology(units); } - return chunkTextForPlan({ - text: params.text, - limit: params.textLimit, - chunker: params.chunker, - formatting: params.formatting, - }).map(planChunkedTextUnit); + return withDeliveryTopology( + chunkTextForPlan({ + text: params.text, + limit: params.textLimit, + chunker: params.chunker, + formatting: params.formatting, + }).map(planChunkedTextUnit), + ); } /** @@ -180,6 +192,7 @@ export function planOutboundMediaMessageUnits(params: { overrides: OutboundMessageSendOverrides; consumeReplyTo?: PlanReplyToConsumption; }): OutboundMessageUnit[] { + const deliveryPartCount = params.mediaUrls.length; return params.mediaUrls.map((mediaUrl, index) => ({ kind: "media" as const, mediaUrl, @@ -187,6 +200,7 @@ export function planOutboundMediaMessageUnits(params: { overrides: { ...withPlannedReplyTo(params.overrides, params.consumeReplyTo), deliveryPartIndex: index, + deliveryPartCount, }, })); }