fix(matrix): retain previews when replacement delivery fails

docs/channels/matrix.md

@@ -234,7 +234,7 @@ The full config accepts `{ mode, chunkMode, block, preview, progress }`:
 Notes:
 
 - If a preview grows past Matrix's per-event size limit, OpenClaw stops preview streaming and falls back to final-only delivery.
-- Media replies always send attachments normally; if a stale preview cannot be reused safely, OpenClaw redacts it before sending the final media reply.
+- Media replies always send attachments normally. If a visible preview cannot be reused safely, OpenClaw keeps it until the complete replacement is confirmed and then redacts it. If replacement delivery fails, is partial, or produces no visible event, the preview remains visible.
 - Tool-progress preview updates are on by default when preview streaming is active. Set `streaming.preview.toolProgress: false` to keep preview edits for answer text but leave tool progress on the normal delivery path.
 - Preview edits cost extra Matrix API calls. Leave `streaming.mode: "off"` for the most conservative rate-limit profile.
 - Legacy scalar/boolean `streaming` values and the flat `blockStreaming` / `chunkMode` keys are rewritten to this nested shape by `openclaw doctor --fix`.

extensions/matrix/src/matrix/monitor/handler-draft-controller.ts

@@ -38,7 +38,8 @@ export async function createMatrixDraftController(params: {
     client,
     logVerboseMessage,
   } = params;
-  let draftConsumed = false;
+  type DraftDisposition = "active" | "retained" | "consumed";
+  let draftDisposition: DraftDisposition = "active";
 
   const draftStreamingEnabled = streaming !== "off";
   const quietDraftStreaming = streaming === "quiet" || streaming === "progress";
@@ -238,7 +239,7 @@ export async function createMatrixDraftController(params: {
   const resetDraftDeliveryState = async () => {
     await draftStream?.discardPending();
     draftStream?.reset();
-    draftConsumed = false;
+    draftDisposition = "active";
     currentDraftMessageGeneration = 0;
     currentDraftBlockOffset = 0;
     latestDraftFullText = "";
@@ -259,12 +260,15 @@ export async function createMatrixDraftController(params: {
     resetPreviewToolProgress,
     resetDraftDeliveryState,
     updateDraftFromLatestFullText,
-    isDraftConsumed: () => draftConsumed,
+    draftDisposition: () => draftDisposition,
-    markDraftConsumed: () => {
+    beginDraftGeneration: () => {
-      draftConsumed = true;
+      draftDisposition = "active";
     },
-    clearDraftConsumed: () => {
+    markDraftConsumed: () => {
-      draftConsumed = false;
+      draftDisposition = "consumed";
+    },
+    markDraftRetained: () => {
+      draftDisposition = "retained";
     },
     currentReplyToId: () => currentDraftReplyToId,
     setCurrentReplyToId: (replyToId: string | undefined) => {

extensions/matrix/src/matrix/monitor/handler-reply-dispatcher.ts

@@ -81,6 +81,15 @@ export function createMatrixReplyDispatcher(config: {
   const hasRepliedRef = { value: false };
   let finalReplyDeliveryFailed = false;
   let nonFinalReplyDeliveryFailed = false;
+  const beginNextBlockDraft = () => {
+    // Each block owns a new draft generation; prior retained/consumed state must not
+    // suppress settlement or cleanup for the next provider-visible event.
+    draftController.beginDraftGeneration();
+    draftController.advanceDraftBlockBoundary({ fallbackToLatestEnd: true });
+    draftStream?.reset();
+    draftController.resetReplyToIdForNextBlock();
+    draftController.updateDraftFromLatestFullText();
+  };
 
   const dispatcherOptions = {
     ...prefixOptions,
@@ -90,11 +99,7 @@ export function createMatrixReplyDispatcher(config: {
         result: MatrixReplyDeliveryResult,
       ): Promise<MatrixReplyDeliveryResult> => {
         if (info.kind === "block") {
-          draftController.clearDraftConsumed();
+          beginNextBlockDraft();
-          draftController.advanceDraftBlockBoundary({ fallbackToLatestEnd: true });
-          draftStream?.reset();
-          draftController.resetReplyToIdForNextBlock();
-          draftController.updateDraftFromLatestFullText();
 
           // Re-assert typing so the user still sees the indicator while
           // the next block generates.
@@ -122,16 +127,30 @@ export function createMatrixReplyDispatcher(config: {
           content,
         };
       };
-      const createSurvivingDraftDelivery = (
+      const settleDraftReplacement = async (params: {
-        id: string,
+        draftEventId: string;
-        redacted: boolean,
+        draftContent: string;
-      ): MatrixReplyDeliveryResult => {
+        deliver: () => Promise<MatrixReplyDeliveryResult>;
-        const content = redacted ? undefined : draftStream?.content();
+      }): Promise<MatrixReplyDeliveryResult> => {
-        return content
+        const draftDelivery = createDraftDeliveryResult(params.draftEventId, params.draftContent);
-          ? // Failed redaction leaves an accepted provider event visible. Preserve it so
+        let replacement: MatrixReplyDeliveryResult;
-            // settlement and retries cannot mistake a partial delivery for total failure.
+        try {
-            createDraftDeliveryResult(id, content)
+          replacement = await params.deliver();
-          : mergeMatrixReplyDeliveryResults([]);
+        } catch (error: unknown) {
+          draftController.markDraftRetained();
+          throw toMatrixPartialDeliveryError(error, [draftDelivery]);
+        }
+        if (!replacement.visibleReplySent) {
+          draftController.markDraftRetained();
+          return draftDelivery;
+        }
+        const draftRedacted = await redactMatrixDraftEvent(client, roomId, params.draftEventId);
+        if (!draftRedacted) {
+          draftController.markDraftRetained();
+          return mergeMatrixReplyDeliveryResults([draftDelivery, replacement]);
+        }
+        draftController.markDraftConsumed();
+        return replacement;
       };
       if (draftStream && info.kind !== "tool" && !payload.isCompactionNotice) {
         const hasMedia = Boolean(payload.mediaUrl) || (payload.mediaUrls?.length ?? 0) > 0;
@@ -143,7 +162,7 @@ export function createMatrixReplyDispatcher(config: {
             ? { ...payload, text: ttsSupplement.spokenText }
             : payload;
 
-        if (draftController.isDraftConsumed()) {
+        if (draftController.draftDisposition() !== "active") {
           await draftStream.discardPending();
           return await completeDelivery(
             await deliverMatrixReplies({
@@ -265,11 +284,11 @@ export function createMatrixReplyDispatcher(config: {
               },
             }),
             deliverNormally: async () => {
-              const draftRedacted = await redactMatrixDraftEvent(client, roomId, draftEventId);
+              fallbackResult = await settleDraftReplacement({
-              const survivingDraft = createSurvivingDraftDelivery(draftEventId, draftRedacted);
+                draftEventId,
-              let deliveredFallback: MatrixReplyDeliveryResult;
+                draftContent: draftStream.content() ?? preparedFinalPreviewContent,
-              try {
+                deliver: async () =>
-                deliveredFallback = await deliverMatrixReplies({
+                  await deliverMatrixReplies({
                     cfg,
                     replies: [fallbackPayload],
                     roomId,
@@ -283,15 +302,14 @@ export function createMatrixReplyDispatcher(config: {
                     accountId,
                     mediaLocalRoots,
                     tableMode,
+                  }),
               });
-              } catch (error: unknown) {
-                throw toMatrixPartialDeliveryError(error, [survivingDraft]);
-              }
-              fallbackResult = mergeMatrixReplyDeliveryResults([survivingDraft, deliveredFallback]);
               return fallbackResult.visibleReplySent;
             },
           });
+          if (previewResult.kind === "preview-finalized") {
             draftController.markDraftConsumed();
+          }
           const settledResult =
             previewResult.kind === "preview-finalized" && previewResult.liveState?.receipt
               ? createDraftDeliveryResult(
@@ -351,9 +369,6 @@ export function createMatrixReplyDispatcher(config: {
           }
           const reusesDraftAsFinalText = Boolean(payloadText?.trim()) && textEditOk;
           const draftContent = draftStream.content();
-          const draftRedacted = reusesDraftAsFinalText
-            ? false
-            : await redactMatrixDraftEvent(client, roomId, draftEventId);
           const mediaPayload =
             ttsSupplement && reusesDraftAsFinalText
               ? buildTtsSupplementMediaPayload(payload)
@@ -370,12 +385,11 @@ export function createMatrixReplyDispatcher(config: {
           const previewDelivery =
             reusesDraftAsFinalText && providerDraftContent
               ? createDraftDeliveryResult(draftEventId, providerDraftContent)
-              : !draftRedacted && draftContent
+              : draftContent
                 ? createDraftDeliveryResult(draftEventId, draftContent)
                 : mergeMatrixReplyDeliveryResults([]);
-          let mediaDelivery: MatrixReplyDeliveryResult;
+          const deliverMedia = async () =>
-          try {
+            await deliverMatrixReplies({
-            mediaDelivery = await deliverMatrixReplies({
               cfg,
               replies: [mediaPayload],
               roomId,
@@ -390,31 +404,37 @@ export function createMatrixReplyDispatcher(config: {
               mediaLocalRoots,
               tableMode,
             });
+          if (reusesDraftAsFinalText) {
+            draftController.markDraftConsumed();
+            let mediaDelivery: MatrixReplyDeliveryResult;
+            try {
+              mediaDelivery = await deliverMedia();
             } catch (error: unknown) {
               throw toMatrixPartialDeliveryError(error, [previewDelivery]);
             }
-          draftController.markDraftConsumed();
             return await completeDelivery(
               mergeMatrixReplyDeliveryResults([previewDelivery, mediaDelivery]),
             );
           }
+          if (draftContent) {
+            return await completeDelivery(
+              await settleDraftReplacement({
+                draftEventId,
+                draftContent,
+                deliver: deliverMedia,
+              }),
+            );
+          }
+          return await completeDelivery(await deliverMedia());
+        }
         const shouldRedactDraft =
           Boolean(draftEventId) &&
           (payload.isError ||
             payloadReplyMismatch ||
             mustDeliverFinalNormally ||
             draftFinalTextNeedsNormalMentionDelivery);
-        const draftRedacted =
+        const deliverFallback = async () =>
-          shouldRedactDraft && draftEventId
+          await deliverMatrixReplies({
-            ? await redactMatrixDraftEvent(client, roomId, draftEventId)
-            : false;
-        const survivingDraft =
-          shouldRedactDraft && draftEventId
-            ? createSurvivingDraftDelivery(draftEventId, draftRedacted)
-            : mergeMatrixReplyDeliveryResults([]);
-        let deliveredFallback: MatrixReplyDeliveryResult;
-        try {
-          deliveredFallback = await deliverMatrixReplies({
             cfg,
             replies: [fallbackPayload],
             roomId,
@@ -429,16 +449,18 @@ export function createMatrixReplyDispatcher(config: {
             mediaLocalRoots,
             tableMode,
           });
-        } catch (error: unknown) {
+        const draftContent = draftStream.content();
-          throw toMatrixPartialDeliveryError(error, [survivingDraft]);
+        if (shouldRedactDraft && draftEventId && draftContent) {
-        }
-        if (shouldRedactDraft || deliveredFallback.visibleReplySent) {
-          draftController.markDraftConsumed();
-        }
           return await completeDelivery(
-          mergeMatrixReplyDeliveryResults([survivingDraft, deliveredFallback]),
+            await settleDraftReplacement({
+              draftEventId,
+              draftContent,
+              deliver: deliverFallback,
+            }),
           );
         }
+        return await completeDelivery(await deliverFallback());
+      }
       return await completeDelivery(
         await deliverMatrixReplies({
           cfg,
@@ -464,7 +486,7 @@ export function createMatrixReplyDispatcher(config: {
         nonFinalReplyDeliveryFailed = true;
       }
       if (info.kind === "block") {
-        draftController.advanceDraftBlockBoundary({ fallbackToLatestEnd: true });
+        beginNextBlockDraft();
       }
       runtime.error?.(`matrix ${info.kind} reply failed: ${String(err)}`);
     },

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

@@ -1,6 +1,7 @@
 import fs from "node:fs";
 import os from "node:os";
 import path from "node:path";
+import { createChannelPartialDeliveryError } from "openclaw/plugin-sdk/channel-inbound";
 import { MAX_DATE_TIMESTAMP_MS } from "openclaw/plugin-sdk/number-runtime";
 import {
   testing as sessionBindingTesting,
@@ -2883,6 +2884,7 @@ describe("matrix monitor handler draft streaming", () => {
       spokenText?: string;
       ttsSupplement?: { spokenText: string; visibleTextAlreadyDelivered?: boolean };
       isCompactionNotice?: boolean;
+      isError?: boolean;
       replyToId?: string;
     },
     info: { kind: string },
@@ -2956,6 +2958,7 @@ describe("matrix monitor handler draft streaming", () => {
     accountConfig?: import("../../types.js").MatrixConfig;
   }) {
     let capturedDeliver: DeliverFn | undefined;
+    let capturedOnError: ((error: unknown, info: { kind: string }) => void) | undefined;
     let capturedReplyOpts: ReplyOpts | undefined;
     let resolveCaptured: (() => void) | undefined;
     const captured = new Promise<void>((resolve) => {
@@ -2992,6 +2995,7 @@ describe("matrix monitor handler draft streaming", () => {
       logVerboseMessage,
       createReplyDispatcherWithTyping: (params: Record<string, unknown> | undefined) => {
         capturedDeliver = params?.deliver as DeliverFn | undefined;
+        capturedOnError = params?.onError as typeof capturedOnError;
         notifyCaptured();
         return {
           dispatcher: {
@@ -3021,6 +3025,7 @@ describe("matrix monitor handler draft streaming", () => {
       await captured;
       return {
         deliver: capturedDeliver!,
+        onError: capturedOnError!,
         opts: capturedReplyOpts!,
         // Release the run gate and wait for the handler to finish
         // (including the finally block that stops the draft stream).
@@ -3408,7 +3413,7 @@ describe("matrix monitor handler draft streaming", () => {
     vi.useRealTimers();
   });
 
-  it("replaces Matrix tool-start progress when command output completes", async () => {
+  it("keeps Matrix tool progress free of terminal status text", async () => {
     vi.useFakeTimers();
     const { dispatch } = createStreamingHarness({
       streaming: "progress",
@@ -3437,7 +3442,7 @@ describe("matrix monitor handler draft streaming", () => {
     await vi.advanceTimersByTimeAsync(5_000);
 
     expect(sendSingleTextMessageMatrixMock).toHaveBeenCalledTimes(1);
-    expect(singleTextMessageBody()).toContain("install dependencies");
+    expect(singleTextMessageBody()).toContain("Exec");
 
     await opts.onItemEvent?.({
       itemId: "fc-call-2",
@@ -3463,7 +3468,7 @@ describe("matrix monitor handler draft streaming", () => {
         eventId === "$draft1" && typeof body === "string" && body.includes("completed"),
     );
     expect(completedEdit).toBeUndefined();
-    expect(singleTextMessageBody()).toContain("install dependencies");
+    expect(singleTextMessageBody()).toContain("Exec");
     vi.useRealTimers();
   });
 
@@ -3853,6 +3858,203 @@ describe("matrix monitor handler draft streaming", () => {
     await finish();
   });
 
+  it.each([
+    { branch: "final-edit", payload: { text: "Final text" }, failEdit: true },
+    { branch: "media", payload: { mediaUrl: "https://example.com/image.png" }, failEdit: false },
+    { branch: "generic", payload: { text: "Something failed", isError: true }, failEdit: false },
+  ])("retains a visible draft when $branch replacement throws", async ({ payload, failEdit }) => {
+    const { dispatch, redactEventMock } = createStreamingHarness({ streaming: "partial" });
+    const { deliver, opts, finish } = await dispatch();
+
+    opts.onPartialReply?.({ text: "Visible preview" });
+    await waitForMatrixState(() => {
+      expect(sendSingleTextMessageMatrixMock).toHaveBeenCalledTimes(1);
+    });
+    if (failEdit) {
+      editMessageMatrixMock.mockRejectedValueOnce(new Error("final edit failed"));
+    }
+    deliverMatrixRepliesMock.mockRejectedValueOnce(new Error("replacement failed"));
+
+    const error = await deliver(payload, { kind: "final" }).catch((caught: unknown) => caught);
+
+    expect(error).toMatchObject({
+      code: "CHANNEL_PARTIAL_DELIVERY",
+      deliveryResult: {
+        messageIds: ["$draft1"],
+        visibleReplySent: true,
+        content: "Visible preview",
+      },
+    });
+    expect(redactEventMock).not.toHaveBeenCalled();
+    await finish();
+    expect(redactEventMock).not.toHaveBeenCalled();
+  });
+
+  it.each([
+    { branch: "final-edit", payload: { text: "Final text" }, failEdit: true },
+    { branch: "media", payload: { mediaUrl: "https://example.com/image.png" }, failEdit: false },
+    { branch: "generic", payload: { text: "Something failed", isError: true }, failEdit: false },
+  ])(
+    "retains a visible draft when $branch replacement reports no visible event",
+    async ({ payload, failEdit }) => {
+      const { dispatch, redactEventMock } = createStreamingHarness({ streaming: "partial" });
+      const { deliver, opts, finish } = await dispatch();
+
+      opts.onPartialReply?.({ text: "Visible preview" });
+      await waitForMatrixState(() => {
+        expect(sendSingleTextMessageMatrixMock).toHaveBeenCalledTimes(1);
+      });
+      if (failEdit) {
+        editMessageMatrixMock.mockRejectedValueOnce(new Error("final edit failed"));
+      }
+      deliverMatrixRepliesMock.mockResolvedValueOnce({
+        visibleReplySent: false,
+        suppression: { reason: "no_visible_result" },
+      });
+
+      const result = await deliver(payload, { kind: "final" });
+
+      expect(result).toMatchObject({
+        messageIds: ["$draft1"],
+        visibleReplySent: true,
+        content: "Visible preview",
+      });
+      expect(redactEventMock).not.toHaveBeenCalled();
+      await finish();
+      expect(redactEventMock).not.toHaveBeenCalled();
+    },
+  );
+
+  it.each([
+    { branch: "final-edit", payload: { text: "Final text" }, failEdit: true },
+    { branch: "media", payload: { mediaUrl: "https://example.com/image.png" }, failEdit: false },
+    { branch: "generic", payload: { text: "Something failed", isError: true }, failEdit: false },
+  ])(
+    "redacts a visible draft only after complete $branch replacement",
+    async ({ payload, failEdit }) => {
+      const { dispatch, redactEventMock } = createStreamingHarness({ streaming: "partial" });
+      const { deliver, opts, finish } = await dispatch();
+
+      opts.onPartialReply?.({ text: "Visible preview" });
+      await waitForMatrixState(() => {
+        expect(sendSingleTextMessageMatrixMock).toHaveBeenCalledTimes(1);
+      });
+      if (failEdit) {
+        editMessageMatrixMock.mockRejectedValueOnce(new Error("final edit failed"));
+      }
+
+      const result = await deliver(payload, { kind: "final" });
+
+      expect(result).toMatchObject({ messageIds: ["$reply1"], visibleReplySent: true });
+      expect(deliverMatrixRepliesMock.mock.invocationCallOrder[0]).toBeLessThan(
+        redactEventMock.mock.invocationCallOrder[0]!,
+      );
+      expect(redactEventMock).toHaveBeenCalledExactlyOnceWith("!room:example.org", "$draft1");
+      await finish();
+      expect(redactEventMock).toHaveBeenCalledTimes(1);
+    },
+  );
+
+  it.each([
+    { branch: "final-edit", payload: { text: "Final text" }, failEdit: true },
+    { branch: "media", payload: { mediaUrl: "https://example.com/image.png" }, failEdit: false },
+    { branch: "generic", payload: { text: "Something failed", isError: true }, failEdit: false },
+  ])(
+    "combines a visible draft with accepted $branch replacement prefixes",
+    async ({ payload, failEdit }) => {
+      const { dispatch, redactEventMock } = createStreamingHarness({ streaming: "partial" });
+      const { deliver, opts, finish } = await dispatch();
+
+      opts.onPartialReply?.({ text: "Visible preview" });
+      await waitForMatrixState(() => {
+        expect(sendSingleTextMessageMatrixMock).toHaveBeenCalledTimes(1);
+      });
+      if (failEdit) {
+        editMessageMatrixMock.mockRejectedValueOnce(new Error("final edit failed"));
+      }
+      deliverMatrixRepliesMock.mockRejectedValueOnce(
+        createChannelPartialDeliveryError(new Error("second replacement event failed"), {
+          ...createMockMatrixDeliveryResult("$accepted-prefix", "Accepted prefix"),
+          visibleReplySent: true as const,
+        }),
+      );
+
+      const error = await deliver(payload, { kind: "final" }).catch((caught: unknown) => caught);
+
+      expect(error).toMatchObject({
+        code: "CHANNEL_PARTIAL_DELIVERY",
+        deliveryResult: {
+          messageIds: ["$draft1", "$accepted-prefix"],
+          visibleReplySent: true,
+          content: "Visible preview\nAccepted prefix",
+        },
+      });
+      expect(redactEventMock).not.toHaveBeenCalled();
+      await finish();
+      expect(redactEventMock).not.toHaveBeenCalled();
+    },
+  );
+
+  it("reports both visible events when post-replacement redaction fails", async () => {
+    const { dispatch, redactEventMock } = createStreamingHarness({ streaming: "partial" });
+    const { deliver, opts, finish } = await dispatch();
+
+    opts.onPartialReply?.({ text: "Visible preview" });
+    await waitForMatrixState(() => {
+      expect(sendSingleTextMessageMatrixMock).toHaveBeenCalledTimes(1);
+    });
+    redactEventMock.mockRejectedValueOnce(new Error("redaction failed"));
+
+    const result = await deliver({ text: "Something failed", isError: true }, { kind: "final" });
+
+    expect(result).toMatchObject({
+      messageIds: ["$draft1", "$reply1"],
+      visibleReplySent: true,
+      content: "Visible preview\ndelivered",
+    });
+    await finish();
+    expect(redactEventMock).toHaveBeenCalledTimes(1);
+  });
+
+  it.each([{ branch: "followup" }, { branch: "block" }])(
+    "starts an active draft generation after a retained $branch boundary",
+    async ({ branch }) => {
+      const { dispatch, redactEventMock } = createStreamingHarness({ streaming: "partial" });
+      const { deliver, onError, opts, finish } = await dispatch();
+
+      opts.onPartialReply?.({ text: "Retained preview" });
+      await waitForMatrixState(() => {
+        expect(sendSingleTextMessageMatrixMock).toHaveBeenCalledTimes(1);
+      });
+      deliverMatrixRepliesMock.mockRejectedValueOnce(new Error("replacement failed"));
+      if (branch === "block") {
+        await opts.onBlockReplyQueued?.({ text: "Retained preview" });
+      }
+      await deliver(
+        { text: "Something failed", isError: true },
+        { kind: branch === "block" ? "block" : "final" },
+      ).catch(() => undefined);
+      if (branch === "followup") {
+        await opts.onQueuedFollowupAdmitted?.();
+      } else {
+        onError(new Error("replacement failed"), { kind: "block" });
+        opts.onAssistantMessageStart?.();
+      }
+
+      sendSingleTextMessageMatrixMock.mockResolvedValueOnce({
+        messageId: "$draft2",
+        roomId: "!room",
+      });
+      opts.onPartialReply?.({ text: "Next generation" });
+      await waitForMatrixState(() => {
+        expect(sendSingleTextMessageMatrixMock).toHaveBeenCalledTimes(2);
+      });
+      await finish();
+
+      expect(redactEventMock).toHaveBeenCalledExactlyOnceWith("!room:example.org", "$draft2");
+    },
+  );
+
   it("falls back with visible text when TTS supplement preview has no event id", async () => {
     const { dispatch, redactEventMock } = createStreamingHarness({
       blockStreamingEnabled: true,

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

@@ -618,7 +618,7 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
       const draftStream = draftControllerRef?.draftStream;
       if (draftStream) {
         const draftEventId = await draftStream.stop().catch(() => undefined);
-        if (draftEventId && draftControllerRef?.isDraftConsumed() !== true) {
+        if (draftEventId && draftControllerRef?.draftDisposition() === "active") {
           await redactMatrixDraftEvent(client, roomId, draftEventId);
         }
       }

extensions/qa-lab/src/live-transports/matrix/scenarios/scenario-runtime-room.ts

@@ -45,6 +45,7 @@ export {
 export {
   runPartialStreamingPreviewScenario,
   runQuietStreamingPreviewScenario,
+  runStreamingReplacementRetentionScenario,
 } from "./scenario-runtime-streaming-preview.js";
 
 export {

extensions/qa-lab/src/live-transports/matrix/scenarios/scenario-runtime-streaming-preview.ts

@@ -33,6 +33,163 @@ export async function runPartialStreamingPreviewScenario(context: MatrixQaScenar
   });
 }
 
+const MATRIX_REPLACEMENT_FAULT_RULE_ID = "matrix-streaming-replacement-failure";
+
+export async function runStreamingReplacementRetentionScenario(
+  context: MatrixQaScenarioContext,
+): Promise<MatrixQaScenarioExecution> {
+  if (!context.installFaultRule) {
+    throw new Error("Matrix streaming replacement QA requires in-place fault injection");
+  }
+  const { client, startSince } = await primeMatrixQaDriverScenarioClient(context);
+  const firstText = `@room ${buildMatrixStreamingPreviewFinalText("MATRIX_QA_RETAINED_DRAFT")}`;
+  const firstToken = firstText.split(" ")[1]!;
+  const firstDriverEventId = await client.sendTextMessage({
+    body: buildMatrixPartialStreamingPrompt(context.sutUserId, firstText),
+    mentionUserIds: [context.sutUserId],
+    roomId: context.roomId,
+  });
+  const firstPreview = await client
+    .waitForRoomEvent({
+      observedEvents: context.observedEvents,
+      predicate: (event) =>
+        event.roomId === context.roomId &&
+        event.sender === context.sutUserId &&
+        isMatrixQaMessageLikeKind(event.kind) &&
+        event.body?.includes(firstToken) === true &&
+        event.body !== firstText,
+      roomId: context.roomId,
+      since: startSince,
+      timeoutMs: context.timeoutMs,
+    })
+    .catch((error: unknown) => {
+      throw new Error("Matrix replacement QA timed out waiting for the first draft", {
+        cause: error,
+      });
+    });
+  const firstDraftEventId = firstPreview.event.replacesEventId ?? firstPreview.event.eventId;
+  const faultRule = context.installFaultRule({
+    id: MATRIX_REPLACEMENT_FAULT_RULE_ID,
+    match: (request) =>
+      request.bearerToken === context.sutAccessToken &&
+      request.path.includes("/send/m.room.message/"),
+    response: () => ({
+      body: { errcode: "M_UNKNOWN", error: "Matrix QA injected replacement failure" },
+      status: 503,
+    }),
+  });
+  let firstWindow;
+  try {
+    firstWindow = await client.waitForOptionalRoomEvent({
+      observedEvents: context.observedEvents,
+      predicate: (event) =>
+        event.roomId === context.roomId &&
+        event.sender === context.sutUserId &&
+        (event.redactsEventId === firstDraftEventId || event.body === firstText),
+      roomId: context.roomId,
+      since: firstPreview.since,
+      timeoutMs: Math.min(8_000, context.timeoutMs),
+    });
+    if (firstWindow.matched) {
+      throw new Error(`Matrix failed replacement did not retain draft ${firstDraftEventId}`);
+    }
+    if (faultRule.hits().length === 0) {
+      throw new Error("Matrix replacement fault rule did not observe a replacement request");
+    }
+  } finally {
+    faultRule.remove();
+  }
+
+  const secondText = `@room ${buildMatrixStreamingPreviewFinalText("MATRIX_QA_SUPERSEDED_DRAFT")}`;
+  const secondToken = secondText.split(" ")[1]!;
+  const secondDriverEventId = await client.sendTextMessage({
+    body: buildMatrixPartialStreamingPrompt(context.sutUserId, secondText),
+    mentionUserIds: [context.sutUserId],
+    roomId: context.roomId,
+  });
+  const secondPreview = await client
+    .waitForRoomEvent({
+      observedEvents: context.observedEvents,
+      predicate: (event) =>
+        event.roomId === context.roomId &&
+        event.sender === context.sutUserId &&
+        isMatrixQaMessageLikeKind(event.kind) &&
+        event.body?.includes(secondToken) === true &&
+        event.eventId !== firstPreview.event.eventId &&
+        event.body !== secondText,
+      roomId: context.roomId,
+      since: firstWindow.since,
+      timeoutMs: context.timeoutMs,
+    })
+    .catch((error: unknown) => {
+      throw new Error("Matrix replacement QA timed out waiting for the second draft", {
+        cause: error,
+      });
+    });
+  const secondDraftEventId = secondPreview.event.replacesEventId ?? secondPreview.event.eventId;
+  const secondReply = await client
+    .waitForRoomEvent({
+      observedEvents: context.observedEvents,
+      predicate: (event) =>
+        event.roomId === context.roomId &&
+        event.sender === context.sutUserId &&
+        isMatrixQaMessageLikeKind(event.kind) &&
+        event.eventId !== secondDraftEventId &&
+        event.replacesEventId === undefined,
+      roomId: context.roomId,
+      since: secondPreview.since,
+      timeoutMs: context.timeoutMs,
+    })
+    .catch((error: unknown) => {
+      throw new Error("Matrix replacement QA timed out waiting for the healthy replacement", {
+        cause: error,
+      });
+    });
+  const secondRedaction = await client
+    .waitForRoomEvent({
+      observedEvents: context.observedEvents,
+      predicate: (event) =>
+        event.roomId === context.roomId &&
+        event.sender === context.sutUserId &&
+        event.kind === "redaction",
+      roomId: context.roomId,
+      since: secondReply.since,
+      timeoutMs: context.timeoutMs,
+    })
+    .catch((error: unknown) => {
+      throw new Error("Matrix replacement QA timed out waiting for post-replacement redaction", {
+        cause: error,
+      });
+    });
+  if (secondRedaction.event.redactsEventId === firstDraftEventId) {
+    throw new Error("Matrix healthy replacement redacted the retained first-generation draft");
+  }
+  advanceMatrixQaActorCursor({
+    actorId: "driver",
+    syncState: context.syncState,
+    nextSince: secondRedaction.since,
+    startSince,
+  });
+  return {
+    artifacts: {
+      faultHitCount: faultRule.hits().length,
+      faultRuleId: MATRIX_REPLACEMENT_FAULT_RULE_ID,
+      firstDriverEventId,
+      previewEventId: firstDraftEventId,
+      redactionEventId: secondRedaction.event.eventId,
+      secondDriverEventId,
+      secondReply: buildMatrixReplyArtifact(secondReply.event, secondText),
+    },
+    details: [
+      `retained draft event: ${firstDraftEventId}`,
+      `replacement fault hits: ${faultRule.hits().length}`,
+      `second draft event: ${secondDraftEventId}`,
+      `second replacement event: ${secondReply.event.eventId}`,
+      `second redaction event: ${secondRedaction.event.eventId}`,
+    ].join("\n"),
+  } satisfies MatrixQaScenarioExecution;
+}
+
 function buildMatrixStreamingPreviewFinalText(prefix: string) {
   const token = `${prefix}_${randomUUID().slice(0, 8).toUpperCase()}`;
   return [

extensions/qa-lab/src/scenario-catalog-channels.test.ts

@@ -27,7 +27,7 @@ describe("qa scenario catalog channel contracts", () => {
       (scenario) => scenario.execution.flowKind === "module",
     );
 
-    expect(moduleFlows).toHaveLength(143);
+    expect(moduleFlows).toHaveLength(144);
     expect(moduleFlows.every((scenario) => scenario.execution.flow)).toBe(true);
   });
 

qa/scenarios/channels/matrix-streaming-replacement-retention.yaml

@@ -0,0 +1,24 @@
+title: Matrix streaming replacement retention
+scenario:
+  id: matrix-streaming-replacement-retention
+  surface: channels
+  coverage:
+    primary:
+      - matrix.conversation-routing-and-delivery
+  docsRefs:
+    - docs/channels/matrix.md
+  execution:
+    kind: flow
+    channel: matrix
+    timeoutMs: 90000
+    retryCount: 0
+    config:
+      matrixConfigOverrides:
+        streaming: partial
+        textChunkLimit: 160
+
+flow:
+  module: ./live-transports/matrix/scenarios/scenario-runtime-room.js
+  call: runStreamingReplacementRetentionScenario
+  args:
+    - expr: "scenarioContext"