fix(matrix): reconcile durable sends after response loss

2026-08-12 21:53:00 -06:00 · 2026-07-31 23:46:06 -05:00
parent f5a8cb02ea
commit 07b7d6446c
31 changed files with 1898 additions and 233 deletions
@@ -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

@@ -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
@@ -117,17 +117,22 @@ function buildMatrixProfileToolSchema(): NonNullable<ChannelMessageToolDiscovery
  };
 }

+function resolveMatrixActionAccount(params: { cfg: CoreConfig; accountId?: string | null }) {
+  if (!params.accountId && requiresExplicitMatrixDefaultAccount(params.cfg)) {
+    return null;
+  }
+  const account = resolveMatrixAccount({
+    cfg: params.cfg,
+    accountId: params.accountId ?? resolveDefaultMatrixAccountId(params.cfg),
+  });
+  return account.enabled && account.configured ? account : null;
+}
+
 export const matrixMessageActions: ChannelMessageActionAdapter = {
  describeMessageTool: ({ cfg, accountId, senderIsOwner }) => {
    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;
@@ -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<typeof matrixChannelRuntime>;
+}) {
+  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;
+}
@@ -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");
+        },
      },
    });
  });
@@ -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,
 };
@@ -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<ResolvedMatrixAccount, MatrixProbe> =
@@ -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<void>,
+    ) => {
+      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<unknown>,
+  ) => 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();
+  });
+});
@@ -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<Record<string, never>>({
+    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<MessageReceiptPartKind>([
+  "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<MatrixDeliveryPlan>;
+  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, "transactionId">[];
+}): 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<MatrixDeliveryPlan | null> {
+  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<MatrixDeliveryPlan> {
+  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<MatrixDeliveryPlan[]> {
+  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<MessageReceipt["parts"]>[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<string> {
+  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<ChannelMessageUnknownSendReconciliationResult> {
+  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<void> {
+  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)));
+}
@@ -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("{}", {
@@ -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<void>;
+
+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<void> | 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<string> | null = null;
+  private readonly messageWireDispatchGuards = new Map<string, MatrixMessageWireDispatchGuard>();

  readonly dms = {
    update: async (): Promise<boolean> => {
@@ -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<T>(params: {
+    transactionId?: string;
+    guard?: MatrixMessageWireDispatchGuard;
+    run: () => Promise<T>;
+  }): Promise<T> {
+    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<TEvent extends keyof MatrixClientEventMap>(
    eventName: TEvent,
    listener: (...args: MatrixClientEventMap[TEvent]) => void,
@@ -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<string> {
+    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<string[]> {
    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<string> {
+  async sendMessage(
+    roomId: string,
+    content: MessageEventContent,
+    transactionId?: string,
+    beforeWireDispatch?: (dispatch: MatrixMessageWireDispatch) => Promise<void>,
+  ): Promise<string> {
    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,
@@ -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<void>,
+      ) => {
+        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();
@@ -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<MatrixPreparedEvent, "transactionId">[] = [];
+        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,
        }),
@@ -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<void>;
  /** 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. */
@@ -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 {
@@ -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<PluginRuntime["state"]>,
-    "openKeyedStore" | "openSyncKeyedStore" | "resolveStateDir"
+    "openBlobStore" | "openKeyedStore" | "openSyncKeyedStore" | "resolveStateDir"
  >;
 };

@@ -89,6 +93,11 @@ export function installMatrixTestRuntime(options: MatrixTestRuntimeOptions = {})
    ...(logging ? { logging } : {}),
    state: {
      resolveStateDir: defaultStateDirResolver,
+      openBlobStore: (<T>(storeOptions: OpenBlobStoreOptions) =>
+        createPluginBlobStoreForTests<T>("matrix", storeOptions, {
+          ...process.env,
+          OPENCLAW_STATE_DIR: defaultStateDirResolver(process.env, osHomedirForTest),
+        })) as PluginRuntime["state"]["openBlobStore"],
      openKeyedStore: (<T>(storeOptions: OpenKeyedStoreOptions) =>
        createPluginStateKeyedStoreForTests<T>("matrix", {
          ...storeOptions,
@@ -183,6 +183,8 @@ export type ChannelMessageSendTextContext<TConfig = OpenClawConfig> = {
  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>
    | ChannelMessageUnknownSendReconciliationResult
    | null;
+  /** Cleanup after core authoritatively retires an ambiguous send as failed. */
+  afterUnknownSendTerminal?: (ctx: ChannelMessageUnknownSendContext) => Promise<void> | void;
 };

 /** Live-message feature key declared by adapters that support preview or streaming behavior. */
@@ -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. */
@@ -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);
  });

@@ -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<void>;
  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);
@@ -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,
@@ -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
@@ -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";
@@ -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.
@@ -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({
@@ -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}`);
@@ -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<void> {
+  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<void> {
+  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<ChannelMessageUnknownSendReconciliationResult, { status: "sent" }>,
@@ -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<boolean> {
  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,
        );
@@ -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 () => {
@@ -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: "<b>bold</b>",
-        overrides: { formatting: { parseMode: "HTML" }, deliveryPartIndex: 0 },
+        overrides: {
+          formatting: { parseMode: "HTML" },
+          deliveryPartIndex: 0,
+          deliveryPartCount: 1,
+        },
      },
    ]);
  });
@@ -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,
    },
  }));
 }