Matrix: restore ordered progress delivery with explicit streaming modes (#59266)

Merged via squash.

Prepared head SHA: 523623b7e1
Co-authored-by: gumadeiras <5599352+gumadeiras@users.noreply.github.com>
Co-authored-by: gumadeiras <5599352+gumadeiras@users.noreply.github.com>
Reviewed-by: @gumadeiras

extensions/matrix/src/config-schema.ts

@@ -65,6 +65,7 @@ export const MatrixConfigSchema = z.object({
   allowlistOnly: z.boolean().optional(),
   allowBots: z.union([z.boolean(), z.literal("mentions")]).optional(),
   groupPolicy: GroupPolicySchema.optional(),
+  blockStreaming: z.boolean().optional(),
   streaming: z.union([z.enum(["partial", "off"]), z.boolean()]).optional(),
   replyToMode: z.enum(["off", "first", "all"]).optional(),
   threadReplies: z.enum(["off", "inbound", "always"]).optional(),

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

@@ -32,6 +32,7 @@ type MatrixHandlerTestHarnessOptions = {
   threadReplies?: "off" | "inbound" | "always";
   dmThreadReplies?: "off" | "inbound" | "always";
   streaming?: "partial" | "off";
+  blockStreamingEnabled?: boolean;
   dmEnabled?: boolean;
   dmPolicy?: "pairing" | "allowlist" | "open" | "disabled";
   textLimit?: number;
@@ -214,6 +215,7 @@ export function createMatrixHandlerTestHarness(
     threadReplies: options.threadReplies ?? "inbound",
     dmThreadReplies: options.dmThreadReplies,
     streaming: options.streaming ?? "off",
+    blockStreamingEnabled: options.blockStreamingEnabled ?? false,
     dmEnabled: options.dmEnabled ?? true,
     dmPolicy: options.dmPolicy ?? "open",
     textLimit: options.textLimit ?? 8_000,

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

@@ -937,6 +937,7 @@ describe("matrix monitor handler pairing account scope", () => {
       replyToMode: "off",
       threadReplies: "inbound",
       streaming: "off",
+      blockStreamingEnabled: false,
       dmEnabled: true,
       dmPolicy: "open",
       textLimit: 8_000,
@@ -1849,3 +1850,91 @@ describe("matrix monitor handler draft streaming", () => {
     await finish();
   });
 });
+
+describe("matrix monitor handler block streaming config", () => {
+  it("keeps final-only delivery when draft streaming is off by default", async () => {
+    let capturedDisableBlockStreaming: boolean | undefined;
+
+    const { handler } = createMatrixHandlerTestHarness({
+      streaming: "off",
+      dispatchReplyFromConfig: vi.fn(
+        async (args: { replyOptions?: { disableBlockStreaming?: boolean } }) => {
+          capturedDisableBlockStreaming = args.replyOptions?.disableBlockStreaming;
+          return { queuedFinal: false, counts: { final: 0, block: 0, tool: 0 } };
+        },
+      ) as never,
+    });
+
+    await handler(
+      "!room:example.org",
+      createMatrixTextMessageEvent({ eventId: "$msg1", body: "hello" }),
+    );
+
+    expect(capturedDisableBlockStreaming).toBe(true);
+  });
+
+  it("disables shared block streaming when draft streaming is partial", async () => {
+    let capturedDisableBlockStreaming: boolean | undefined;
+
+    const { handler } = createMatrixHandlerTestHarness({
+      streaming: "partial",
+      dispatchReplyFromConfig: vi.fn(
+        async (args: { replyOptions?: { disableBlockStreaming?: boolean } }) => {
+          capturedDisableBlockStreaming = args.replyOptions?.disableBlockStreaming;
+          return { queuedFinal: false, counts: { final: 0, block: 0, tool: 0 } };
+        },
+      ) as never,
+    });
+
+    await handler(
+      "!room:example.org",
+      createMatrixTextMessageEvent({ eventId: "$msg1", body: "hello" }),
+    );
+
+    expect(capturedDisableBlockStreaming).toBe(true);
+  });
+
+  it("keeps draft streaming authoritative when partial and block streaming are both enabled", async () => {
+    let capturedDisableBlockStreaming: boolean | undefined;
+
+    const { handler } = createMatrixHandlerTestHarness({
+      streaming: "partial",
+      blockStreamingEnabled: true,
+      dispatchReplyFromConfig: vi.fn(
+        async (args: { replyOptions?: { disableBlockStreaming?: boolean } }) => {
+          capturedDisableBlockStreaming = args.replyOptions?.disableBlockStreaming;
+          return { queuedFinal: false, counts: { final: 0, block: 0, tool: 0 } };
+        },
+      ) as never,
+    });
+
+    await handler(
+      "!room:example.org",
+      createMatrixTextMessageEvent({ eventId: "$msg1", body: "hello" }),
+    );
+
+    expect(capturedDisableBlockStreaming).toBe(true);
+  });
+
+  it("uses shared block streaming when explicitly enabled for Matrix", async () => {
+    let capturedDisableBlockStreaming: boolean | undefined;
+
+    const { handler } = createMatrixHandlerTestHarness({
+      streaming: "off",
+      blockStreamingEnabled: true,
+      dispatchReplyFromConfig: vi.fn(
+        async (args: { replyOptions?: { disableBlockStreaming?: boolean } }) => {
+          capturedDisableBlockStreaming = args.replyOptions?.disableBlockStreaming;
+          return { queuedFinal: false, counts: { final: 0, block: 0, tool: 0 } };
+        },
+      ) as never,
+    });
+
+    await handler(
+      "!room:example.org",
+      createMatrixTextMessageEvent({ eventId: "$msg1", body: "hello" }),
+    );
+
+    expect(capturedDisableBlockStreaming).toBe(false);
+  });
+});

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

@@ -84,6 +84,7 @@ export type MatrixMonitorHandlerParams = {
   /** DM-specific threadReplies override. Falls back to threadReplies when absent. */
   dmThreadReplies?: "off" | "inbound" | "always";
   streaming: "partial" | "off";
+  blockStreamingEnabled: boolean;
   dmEnabled: boolean;
   dmPolicy: "open" | "pairing" | "allowlist" | "disabled";
   textLimit: number;
@@ -201,6 +202,7 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
     threadReplies,
     dmThreadReplies,
     streaming,
+    blockStreamingEnabled,
     dmEnabled,
     dmPolicy,
     textLimit,
@@ -1127,10 +1129,10 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
           });
         },
       });
-      const streamingEnabled = streaming === "partial";
+      const draftStreamingEnabled = streaming === "partial";
       const draftReplyToId = replyToMode !== "off" && !threadTarget ? _messageId : undefined;
       let currentDraftReplyToId = draftReplyToId;
-      const draftStream = streamingEnabled
+      const draftStream = draftStreamingEnabled
         ? createMatrixDraftStream({
             roomId,
             client,
@@ -1350,9 +1352,11 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
               replyOptions: {
                 ...replyOptions,
                 skillFilter: roomConfig?.skills,
-                // When streaming is active, disable block streaming — draft
-                // streaming replaces it with edit-in-place updates.
-                disableBlockStreaming: streamingEnabled ? true : undefined,
+                // Matrix expects explicit assistant progress updates as
+                // separate messages only when block streaming is explicitly
+                // enabled. Partial draft streaming still disables the shared
+                // block pipeline so draft edits do not double-send.
+                disableBlockStreaming: draftStream ? true : !blockStreamingEnabled,
                 onPartialReply: draftStream
                   ? (payload) => {
                       const fullText = payload.text ?? "";

extensions/matrix/src/matrix/monitor/index.ts

@@ -210,6 +210,7 @@ export async function monitorMatrixProvider(opts: MonitorMatrixOpts = {}): Promi
   const mediaMaxBytes = Math.max(1, mediaMaxMb) * 1024 * 1024;
   const streaming: "partial" | "off" =
     accountConfig.streaming === true || accountConfig.streaming === "partial" ? "partial" : "off";
+  const blockStreamingEnabled = accountConfig.blockStreaming === true;
   const startupMs = Date.now();
   const startupGraceMs = 0;
   // Cold starts should ignore old room history, but once we have a persisted
@@ -265,6 +266,7 @@ export async function monitorMatrixProvider(opts: MonitorMatrixOpts = {}): Promi
     threadReplies,
     dmThreadReplies,
     streaming,
+    blockStreamingEnabled,
     dmEnabled,
     dmPolicy,
     textLimit,

extensions/matrix/src/setup-config.ts

@@ -39,6 +39,7 @@ const MATRIX_SINGLE_ACCOUNT_KEYS_TO_MOVE = new Set([
   "encryption",
   "allowlistOnly",
   "allowBots",
+  "blockStreaming",
   "replyToMode",
   "threadReplies",
   "textChunkLimit",
@@ -59,6 +60,9 @@ const MATRIX_SINGLE_ACCOUNT_KEYS_TO_MOVE = new Set([
   "actions",
 ]);
 const MATRIX_NAMED_ACCOUNT_PROMOTION_KEYS = new Set([
+  // When named accounts already exist, only move auth/bootstrap fields into the
+  // promoted account. Delivery-policy fields stay at the top level so they
+  // remain shared inherited defaults for every account.
   "name",
   "homeserver",
   "userId",

extensions/matrix/src/setup-core.test.ts

@@ -104,4 +104,45 @@ describe("matrixSetupAdapter", () => {
       proxy: "http://127.0.0.1:7890",
     });
   });
+
+  it("keeps top-level block streaming as a shared default when named accounts already exist", () => {
+    const cfg = {
+      channels: {
+        matrix: {
+          homeserver: "https://matrix.example.org",
+          userId: "@default:example.org",
+          accessToken: "default-token",
+          blockStreaming: true,
+          accounts: {
+            support: {
+              homeserver: "https://matrix.example.org",
+              userId: "@support:example.org",
+              accessToken: "support-token",
+            },
+          },
+        },
+      },
+    } as CoreConfig;
+
+    const next = matrixSetupAdapter.applyAccountConfig({
+      cfg,
+      accountId: "ops",
+      input: {
+        name: "Ops",
+        homeserver: "https://matrix.example.org",
+        userId: "@ops:example.org",
+        accessToken: "ops-token",
+      },
+    }) as CoreConfig;
+
+    expect(next.channels?.matrix?.blockStreaming).toBe(true);
+    expect(next.channels?.matrix?.accounts?.ops).toMatchObject({
+      name: "Ops",
+      enabled: true,
+      homeserver: "https://matrix.example.org",
+      userId: "@ops:example.org",
+      accessToken: "ops-token",
+    });
+    expect(next.channels?.matrix?.accounts?.ops?.blockStreaming).toBeUndefined();
+  });
 });

extensions/matrix/src/types.ts

@@ -101,6 +101,13 @@ export type MatrixConfig = {
   allowBots?: boolean | "mentions";
   /** Group message policy (default: allowlist). */
   groupPolicy?: GroupPolicy;
+  /**
+   * Enable shared block-streaming replies for Matrix.
+   *
+   * Default: false. Matrix keeps `streaming: "off"` as final-only delivery
+   * unless block streaming is explicitly enabled.
+   */
+  blockStreaming?: boolean;
   /** Allowlist for group senders (matrix user IDs). */
   groupAllowFrom?: Array<string | number>;
   /** Control reply threading when reply tags are present (off|first|all). */
@@ -149,6 +156,8 @@ export type MatrixConfig = {
    * Streaming mode for Matrix replies.
    * - `"partial"`: edit a single message in place as the model generates text.
    * - `"off"`: deliver the full reply once the model finishes.
+   * - Use `blockStreaming: true` when you want separate progress messages
+   *   while `streaming` remains `"off"`.
    * - `true` maps to `"partial"`, `false` maps to `"off"`.
    * Default: `"off"`.
    */