matrix: only consume drafts after real fallback delivery

extensions/matrix/src/matrix/monitor/handler.test.ts

@@ -48,7 +48,7 @@ vi.mock("../send.js", () => ({
   sendTypingMatrix: vi.fn(async () => {}),
 }));
 
-const deliverMatrixRepliesMock = vi.hoisted(() => vi.fn(async () => {}));
+const deliverMatrixRepliesMock = vi.hoisted(() => vi.fn(async () => true));
 
 vi.mock("./replies.js", () => ({
   deliverMatrixReplies: deliverMatrixRepliesMock,
@@ -2005,7 +2005,7 @@ describe("matrix monitor handler draft streaming", () => {
       .mockReset()
       .mockResolvedValue({ messageId: "$draft1", roomId: "!room" });
     editMessageMatrixMock.mockReset().mockResolvedValue("$edited");
-    deliverMatrixRepliesMock.mockReset().mockResolvedValue(undefined);
+    deliverMatrixRepliesMock.mockReset().mockResolvedValue(true);
 
     const redactEventMock = vi.fn(async () => "$redacted");
 
@@ -2530,7 +2530,7 @@ describe("matrix monitor handler draft streaming", () => {
         .mockReset()
         .mockResolvedValue({ messageId: "$draft1", roomId: "!room" });
       editMessageMatrixMock.mockReset().mockResolvedValue("$edited");
-      deliverMatrixRepliesMock.mockReset().mockResolvedValue(undefined);
+      deliverMatrixRepliesMock.mockReset().mockResolvedValue(true);
       const redactEventMock = vi.fn(async () => "$redacted");
 
       let capturedReplyOpts: ReplyOpts | undefined;
@@ -2588,7 +2588,7 @@ describe("matrix monitor handler draft streaming", () => {
       .mockReset()
       .mockResolvedValue({ messageId: "$draft1", roomId: "!room" });
     editMessageMatrixMock.mockReset().mockResolvedValue("$edited");
-    deliverMatrixRepliesMock.mockReset().mockResolvedValue(undefined);
+    deliverMatrixRepliesMock.mockReset().mockResolvedValue(true);
 
     const redactEventMock = vi.fn(async () => "$redacted");
     let capturedReplyOpts: ReplyOpts | undefined;
@@ -2629,6 +2629,27 @@ describe("matrix monitor handler draft streaming", () => {
     expect(redactEventMock).toHaveBeenCalledWith("!room:example.org", "$draft1");
   });
 
+  it("keeps shutdown cleanup for empty final payloads that send nothing", async () => {
+    const { dispatch, redactEventMock } = createStreamingHarness({ streaming: "partial" });
+    const { deliver, opts, finish } = await dispatch();
+
+    opts.onPartialReply?.({ text: "Partial reply" });
+    await vi.waitFor(() => {
+      expect(sendSingleTextMessageMatrixMock).toHaveBeenCalledTimes(1);
+    });
+
+    deliverMatrixRepliesMock.mockClear();
+    deliverMatrixRepliesMock.mockResolvedValue(false);
+    await deliver({}, { kind: "final" });
+
+    expect(deliverMatrixRepliesMock).toHaveBeenCalledTimes(1);
+    expect(redactEventMock).not.toHaveBeenCalled();
+
+    await finish();
+
+    expect(redactEventMock).toHaveBeenCalledWith("!room:example.org", "$draft1");
+  });
+
   it("skips compaction notices in draft finalization", async () => {
     const { dispatch } = createStreamingHarness();
     const { deliver, opts, finish } = await dispatch();

extensions/matrix/src/matrix/monitor/handler.ts

@@ -1517,10 +1517,12 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
                 });
                 draftConsumed = true;
               } else {
-                if (draftEventId && (payloadReplyMismatch || mustDeliverFinalNormally)) {
+                const draftRedacted =
+                  Boolean(draftEventId) && (payloadReplyMismatch || mustDeliverFinalNormally);
+                if (draftRedacted && draftEventId) {
                   await redactMatrixDraftEvent(client, roomId, draftEventId);
                 }
-                await deliverMatrixReplies({
+                const deliveredFallback = await deliverMatrixReplies({
                   cfg,
                   replies: [payload],
                   roomId,
@@ -1533,7 +1535,7 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
                   mediaLocalRoots,
                   tableMode,
                 });
-                if (draftEventId) {
+                if (draftRedacted || deliveredFallback) {
                   draftConsumed = true;
                 }
               }

extensions/matrix/src/matrix/monitor/replies.ts

@@ -41,7 +41,7 @@ export async function deliverMatrixReplies(params: {
   accountId?: string;
   mediaLocalRoots?: readonly string[];
   tableMode?: MarkdownTableMode;
-}): Promise<void> {
+}): Promise<boolean> {
   const core = getMatrixRuntime();
   const tableMode =
     params.tableMode ??
@@ -56,6 +56,7 @@ export async function deliverMatrixReplies(params: {
     }
   };
   let hasReplied = false;
+  let deliveredAny = false;
   for (const reply of params.replies) {
     if (reply.isReasoning === true || shouldSuppressReasoningReplyText(reply.text)) {
       logVerbose("matrix reply suppressed as reasoning-only");
@@ -102,6 +103,7 @@ export async function deliverMatrixReplies(params: {
           threadId: params.threadId,
           accountId: params.accountId,
         });
+        deliveredAny = true;
         sentTextChunk = true;
       }
       if (replyToIdForReply && !hasReplied && sentTextChunk) {
@@ -123,10 +125,12 @@ export async function deliverMatrixReplies(params: {
         audioAsVoice: reply.audioAsVoice,
         accountId: params.accountId,
       });
+      deliveredAny = true;
       first = false;
     }
     if (replyToIdForReply && !hasReplied) {
       hasReplied = true;
     }
   }
+  return deliveredAny;
 }