fix(qa): isolate Matrix live validation state (#116071)

* fix(qa): isolate Matrix live scenario routing * fix(qa): preserve Matrix mention filenames * fix(qa): reset Matrix transport scenario state * test(qa): isolate stateful Matrix scenario partitions * test(qa): match numbered Matrix isolation * test(qa): reject stale Matrix progress results * fix(qa): bound Matrix scenario prompt ownership * fix(qa): bound mock scenario ownership * test(qa): satisfy Matrix validation gates * refactor(qa): remove stale prompt helper * fix(qa): preserve tool-result ownership after rebase
2026-08-02 04:21:34 +00:00 · 2026-07-29 17:23:03 -04:00
parent fc3f114a93
commit d98e984ec2
36 changed files with 1010 additions and 141 deletions
--- a/extensions/qa-lab/src/live-transports/matrix/adapter.runtime.test.ts
+++ b/extensions/qa-lab/src/live-transports/matrix/adapter.runtime.test.ts
@@ -0,0 +1,111 @@
+import { describe, expect, it, vi } from "vitest";
+import { waitForMatrixQaObserverEvent } from "./adapter.runtime.js";
+import type { MatrixQaRoomObserver } from "./substrate/sync.js";
+
+function makeObserver(
+  waitForOptionalRoomEvent: MatrixQaRoomObserver["waitForOptionalRoomEvent"],
+): MatrixQaRoomObserver {
+  return {
+    prime: vi.fn(async () => "s0"),
+    waitForOptionalRoomEvent,
+    waitForRoomEvent: vi.fn(),
+  };
+}
+
+describe("Matrix QA adapter observer recovery", () => {
+  it("retries a sync failure only during the explicit transport interruption window", async () => {
+    let interrupted = true;
+    const waitForOptionalRoomEvent = vi
+      .fn<MatrixQaRoomObserver["waitForOptionalRoomEvent"]>()
+      .mockRejectedValueOnce(new Error("homeserver unavailable"))
+      .mockResolvedValueOnce({ matched: false, since: "s1" });
+    const sleepImpl = vi.fn(async () => {
+      interrupted = false;
+    });
+
+    await expect(
+      waitForMatrixQaObserverEvent({
+        isExpectedInterruption: () => interrupted,
+        observer: makeObserver(waitForOptionalRoomEvent),
+        predicate: () => true,
+        readInterruptionGeneration: () => 1,
+        roomId: "!room:matrix.test",
+        sleepImpl,
+        timeoutMs: 1_000,
+      }),
+    ).resolves.toEqual({ matched: false, since: "s1" });
+    expect(waitForOptionalRoomEvent).toHaveBeenCalledTimes(2);
+    expect(sleepImpl).toHaveBeenCalledOnce();
+  });
+
+  it("preserves terminal observer failures outside the interruption window", async () => {
+    const failure = new Error("sync failed");
+    const waitForOptionalRoomEvent = vi
+      .fn<MatrixQaRoomObserver["waitForOptionalRoomEvent"]>()
+      .mockRejectedValue(failure);
+    const sleepImpl = vi.fn(async () => undefined);
+
+    await expect(
+      waitForMatrixQaObserverEvent({
+        isExpectedInterruption: () => false,
+        observer: makeObserver(waitForOptionalRoomEvent),
+        predicate: () => true,
+        readInterruptionGeneration: () => 0,
+        roomId: "!room:matrix.test",
+        sleepImpl,
+        timeoutMs: 1_000,
+      }),
+    ).rejects.toBe(failure);
+    expect(waitForOptionalRoomEvent).toHaveBeenCalledOnce();
+    expect(sleepImpl).not.toHaveBeenCalled();
+  });
+
+  it("retries a poll that started during interruption after readiness closes the window", async () => {
+    let interrupted = true;
+    const waitForOptionalRoomEvent = vi
+      .fn<MatrixQaRoomObserver["waitForOptionalRoomEvent"]>()
+      .mockImplementationOnce(async () => {
+        interrupted = false;
+        throw new Error("stale outage response");
+      })
+      .mockResolvedValueOnce({ matched: false, since: "s2" });
+
+    await expect(
+      waitForMatrixQaObserverEvent({
+        isExpectedInterruption: () => interrupted,
+        observer: makeObserver(waitForOptionalRoomEvent),
+        predicate: () => true,
+        readInterruptionGeneration: () => 1,
+        roomId: "!room:matrix.test",
+        sleepImpl: vi.fn(async () => undefined),
+        timeoutMs: 1_000,
+      }),
+    ).resolves.toEqual({ matched: false, since: "s2" });
+    expect(waitForOptionalRoomEvent).toHaveBeenCalledTimes(2);
+  });
+
+  it("retries a poll that spans the complete interruption window", async () => {
+    let interruptionGeneration = 0;
+    const waitForOptionalRoomEvent = vi
+      .fn<MatrixQaRoomObserver["waitForOptionalRoomEvent"]>()
+      .mockImplementationOnce(async () => {
+        // The poll began before restart and rejected only after begin/end both ran.
+        interruptionGeneration = 2;
+        throw new Error("pre-restart poll rejected after recovery");
+      })
+      .mockResolvedValueOnce({ matched: false, since: "s3" });
+
+    await expect(
+      waitForMatrixQaObserverEvent({
+        isExpectedInterruption: () => false,
+        observer: makeObserver(waitForOptionalRoomEvent),
+        predicate: () => true,
+        readInterruptionGeneration: () => interruptionGeneration,
+        roomId: "!room:matrix.test",
+        sleepImpl: vi.fn(async () => undefined),
+        timeoutMs: 1_000,
+      }),
+    ).resolves.toEqual({ matched: false, since: "s3" });
+    expect(waitForOptionalRoomEvent).toHaveBeenCalledTimes(2);
+  });
+});
--- a/extensions/qa-lab/src/live-transports/matrix/adapter.runtime.ts
+++ b/extensions/qa-lab/src/live-transports/matrix/adapter.runtime.ts
@@ -1,6 +1,7 @@
 // Qa Lab plugin module implements Matrix live transport adapter behavior.
 import { randomUUID } from "node:crypto";
 import path from "node:path";
+import { setTimeout as sleep } from "node:timers/promises";
 import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
 import { buildQaTarget } from "openclaw/plugin-sdk/qa-channel-protocol";
 import type { QaRunnerCliRegistration } from "openclaw/plugin-sdk/qa-runner-runtime";
@@ -10,7 +11,11 @@ import { createMatrixQaClient, provisionMatrixQaRoom } from "./substrate/client.
 import { buildMatrixQaConfig } from "./substrate/config.js";
 import type { MatrixQaObservedEvent } from "./substrate/events.js";
 import { startMatrixQaHarness } from "./substrate/harness.runtime.js";
-import { createMatrixQaRoomObserver } from "./substrate/sync.js";
+import {
+  createMatrixQaRoomObserver,
+  type MatrixQaRoomEventWaitResult,
+  type MatrixQaRoomObserver,
+} from "./substrate/sync.js";
 import {
  mergeMatrixQaTopologySpecs,
  resolveMatrixQaRoomObserverRole,
@@ -54,6 +59,44 @@ const MATRIX_SHARED_FLOW_TOPOLOGY = {
  ],
 } satisfies MatrixQaTopologySpec;

+const MATRIX_EXPECTED_INTERRUPTION_RETRY_MS = 250;
+
+export async function waitForMatrixQaObserverEvent(params: {
+  isExpectedInterruption: () => boolean;
+  observer: MatrixQaRoomObserver;
+  predicate: (event: MatrixQaObservedEvent) => boolean;
+  readInterruptionGeneration: () => number;
+  roomId: string;
+  sleepImpl?: (ms: number) => Promise<unknown>;
+  timeoutMs: number;
+}): Promise<MatrixQaRoomEventWaitResult> {
+  const sleepImpl = params.sleepImpl ?? sleep;
+  for (;;) {
+    const expectedInterruptionAtStart = params.isExpectedInterruption();
+    const interruptionGenerationAtStart = params.readInterruptionGeneration();
+    try {
+      return await params.observer.waitForOptionalRoomEvent({
+        predicate: params.predicate,
+        roomId: params.roomId,
+        timeoutMs: params.timeoutMs,
+      });
+    } catch (error) {
+      // The homeserver restart scenario owns this narrow recovery window. The
+      // generation also catches a poll that spans the complete interruption
+      // before rejecting. The observer clears its failed pollPromise in finally,
+      // so the same cursor can safely retry.
+      if (
+        !expectedInterruptionAtStart &&
+        !params.isExpectedInterruption() &&
+        interruptionGenerationAtStart === params.readInterruptionGeneration()
+      ) {
+        throw error;
+      }
+      await sleepImpl(MATRIX_EXPECTED_INTERRUPTION_RETRY_MS);
+    }
+  }
+}
+
 function readMatrixQaScenarioTopology(scenarioId: string): MatrixQaTopologySpec | undefined {
  const value = readQaScenarioExecutionConfig(scenarioId)?.matrixTopology;
  if (!value || typeof value !== "object" || Array.isArray(value)) {
@@ -203,9 +246,17 @@ export async function createMatrixQaTransportAdapter(
  );
  const nativeEventIds = new Map<string, string>();
  const busMessageIds = new Map<string, string>();
+  let expectedTransportInterruption = false;
+  let transportInterruptionGeneration = 0;
  const scenarioEnvironment = createMatrixQaScenarioEnvironment({
    accountId,
    harness,
+    onTransportInterruptionStateChange: (active) => {
+      if (expectedTransportInterruption !== active) {
+        transportInterruptionGeneration += 1;
+      }
+      expectedTransportInterruption = active;
+    },
    observedEvents,
    provisioning,
  });
@@ -215,8 +266,11 @@ export async function createMatrixQaTransportAdapter(
        if (stopped) {
          return;
        }
-        const observed = await observer.waitForOptionalRoomEvent({
+        const observed = await waitForMatrixQaObserverEvent({
+          isExpectedInterruption: () => expectedTransportInterruption,
+          observer,
          predicate: (event) => event.sender === provisioning.sut.userId && Boolean(event.body),
+          readInterruptionGeneration: () => transportInterruptionGeneration,
          roomId,
          timeoutMs: 1_000,
        });
--- a/extensions/qa-lab/src/live-transports/matrix/matrix-scenario-flows.test.ts
+++ b/extensions/qa-lab/src/live-transports/matrix/matrix-scenario-flows.test.ts
@@ -4,6 +4,7 @@ import {
  readQaScenarioById,
  readQaScenarioExecutionConfig,
 } from "../../scenario-catalog.js";
+import { requireFlowScenario } from "../../scenario-catalog.test-utils.js";

 const MATRIX_MENTION_GATE_PRIMARY_SCENARIOS = [
  "matrix-allowbots-default-block",
@@ -16,6 +17,12 @@ const MATRIX_MENTION_GATE_PRIMARY_SCENARIOS = [
  "matrix-mention-metadata-spoof-block",
 ] as const;

+const MATRIX_ISOLATED_ALLOWBOTS_ADMISSION_SCENARIOS = [
+  "matrix-allowbots-mentions-mentioned-room",
+  "matrix-allowbots-room-override-enables-account-off",
+  "matrix-allowbots-true-unmentioned-open-room",
+] as const;
+
 function readModuleBinding(
  scenario: ReturnType<typeof readQaBootstrapScenarioCatalog>["scenarios"][number],
 ) {
@@ -124,6 +131,65 @@ describe("Matrix QA Lab scenario flows", () => {
    });
  });

+  it("isolates scenarios that assert fresh thread and DM session routing", () => {
+    const expectedReasons = {
+      "dm-shared-session": "pristine per-user DM session routing",
+      "thread-isolation": "fresh session boundary",
+      "thread-reply-override": "must not inherit an existing room session",
+    } as const;
+
+    for (const [scenarioId, reason] of Object.entries(expectedReasons)) {
+      const execution = requireFlowScenario(readQaScenarioById(scenarioId)).execution;
+      expect(execution.suiteIsolation, scenarioId).toBe("isolated");
+      expect(execution.isolationReason, scenarioId).toContain(reason);
+    }
+  });
+
+  it("isolates the homeserver restart from the shared Matrix sync streams", () => {
+    expect(readQaScenarioById("matrix-homeserver-restart-resume").execution).toMatchObject({
+      kind: "flow",
+      channel: "matrix",
+      suiteIsolation: "isolated",
+      isolationReason:
+        "Restarts the disposable homeserver process and its active Matrix sync streams.",
+    });
+  });
+
+  it("isolates the DM thread override from session-scope bindings", () => {
+    expect(readQaScenarioById("matrix-dm-thread-reply-override").execution).toMatchObject({
+      kind: "flow",
+      channel: "matrix",
+      suiteIsolation: "isolated",
+      isolationReason:
+        "Asserts fresh DM native-thread routing after session-scope scenarios and cannot inherit their DM session binding.",
+    });
+  });
+
+  it("isolates channel and DM approval fan-out from shared routing state", () => {
+    expect(readQaScenarioById("matrix-approval-channel-target-both").execution).toMatchObject({
+      kind: "flow",
+      channel: "matrix",
+      suiteIsolation: "isolated",
+      isolationReason:
+        "Asserts fresh channel and DM approval fan-out and cannot inherit shared Matrix approval routing state.",
+    });
+  });
+
+  it("isolates only model-driven allowBots admission scenarios", () => {
+    const isolatedScenarioIds = MATRIX_MENTION_GATE_PRIMARY_SCENARIOS.filter(
+      (scenarioId) =>
+        requireFlowScenario(readQaScenarioById(scenarioId)).execution.suiteIsolation === "isolated",
+    );
+
+    expect(isolatedScenarioIds).toEqual(MATRIX_ISOLATED_ALLOWBOTS_ADMISSION_SCENARIOS);
+    for (const scenarioId of MATRIX_ISOLATED_ALLOWBOTS_ADMISSION_SCENARIOS) {
+      expect(
+        requireFlowScenario(readQaScenarioById(scenarioId)).execution.isolationReason,
+        scenarioId,
+      ).toContain("fresh model-driven configured-bot admission");
+    }
+  });
+
  it("runs the allowlist scenario through its config-file reload owner", () => {
    const scenario = catalog.scenarios.find((entry) => entry.id === "matrix-allowlist-hot-reload");
    expect(scenario?.execution.kind).toBe("flow");
--- a/extensions/qa-lab/src/live-transports/matrix/scenarios/scenario-environment.test.ts
+++ b/extensions/qa-lab/src/live-transports/matrix/scenarios/scenario-environment.test.ts
@@ -9,13 +9,91 @@ vi.mock("../substrate/config.js", () => ({ buildMatrixQaConfig }));
 vi.mock("./scenario-runtime-room.js", () => ({ runMatrixQaCanary: vi.fn() }));

 import { createMatrixQaScenarioEnvironment } from "./scenario-environment.js";
+import type { MatrixQaScenarioContext } from "./scenario-runtime-shared.js";

 afterEach(() => {
  vi.useRealTimers();
 });

 describe("matrix scenario environment", () => {
-  it("waits for config restart settle before accepting Matrix readiness", async () => {
+  it("drops actor sync cursors and observers at a scenario boundary", async () => {
+    let configReadCount = 0;
+    const gateway = {
+      baseUrl: "http://127.0.0.1:12345",
+      runtimeEnv: {},
+      tempRoot: "/tmp/matrix-qa",
+      workspaceDir: "/tmp/matrix-qa/workspace",
+      call: vi.fn(async (method: string) => {
+        if (method === "config.get") {
+          configReadCount += 1;
+          const phase = (configReadCount - 1) % 3;
+          if (phase === 0) {
+            return { config: {} };
+          }
+          if (phase === 1) {
+            return { hash: "config-hash" };
+          }
+          return {
+            appliedConfigHash: "config-hash",
+            configRevisionHash: "config-hash",
+            hash: "config-hash",
+          };
+        }
+        if (method === "config.patch") {
+          return { hash: "config-hash", noop: true, ok: true };
+        }
+        if (method === "channels.status") {
+          return {
+            channelAccounts: {
+              matrix: [
+                {
+                  accountId: "sut",
+                  connected: true,
+                  healthState: "healthy",
+                  lastStartAt: 100,
+                  restartPending: false,
+                  running: true,
+                },
+              ],
+            },
+          };
+        }
+        throw new Error(`unexpected gateway method ${method}`);
+      }),
+    };
+    const environment = createMatrixQaScenarioEnvironment({
+      accountId: "sut",
+      harness: { baseUrl: "http://127.0.0.1:8008", recording: {} } as never,
+      observedEvents: [],
+      provisioning: {
+        driver: { accessToken: "fixture", userId: "@driver:test" },
+        observer: { accessToken: "fixture", userId: "@observer:test" },
+        roomId: "!room:test",
+        sut: { accessToken: "fixture", userId: "@sut:test" },
+        topology: { rooms: [] },
+      } as never,
+    });
+    const input = {
+      config: {},
+      gateway,
+      outputDir: "/tmp/matrix-qa/output",
+      timeoutMs: 1_000,
+      waitForConfigRestartSettle: vi.fn(),
+    };
+    const first = await environment.prepareFlow(input);
+    const syncState: MatrixQaScenarioContext["syncState"] = first.scenarioContext.syncState;
+    syncState.driver = "s1";
+    syncState.observer = "s2";
+    first.scenarioContext.syncStreams!.driver = { prime: vi.fn() } as never;
+    first.scenarioContext.syncStreams!.observer = { prime: vi.fn() } as never;
+
+    const second = await environment.prepareFlow(input);
+
+    expect(second.scenarioContext.syncState).toEqual({});
+    expect(second.scenarioContext.syncStreams).toEqual({});
+  });
+
+  it("waits for the changed Matrix account to restart before accepting readiness", async () => {
    vi.useFakeTimers();
    const callOrder: string[] = [];
    let configReadCount = 0;
@@ -45,7 +123,6 @@ describe("matrix scenario environment", () => {
          return {
            hash: "patched-config-hash",
            ok: true,
-            sentinel: { payload: { stats: { requiresRestart: true } } },
          };
        }
        if (method === "channels.status") {
@@ -123,7 +200,11 @@ describe("matrix scenario environment", () => {
    expect(gateway.call).toHaveBeenCalledWith(
      "config.patch",
      expect.objectContaining({
-        replacePaths: ["channels.matrix", "messages"],
+        replacePaths: [
+          "channels.matrix",
+          "channels.matrix.accounts.sut.groupAllowFrom",
+          "messages",
+        ],
      }),
      { timeoutMs: 60_000 },
    );
--- a/extensions/qa-lab/src/live-transports/matrix/scenarios/scenario-environment.ts
+++ b/extensions/qa-lab/src/live-transports/matrix/scenarios/scenario-environment.ts
@@ -1,5 +1,6 @@
 // QA Lab Matrix setup prepares transport state for the shared flow host.
 import { setTimeout as sleep } from "node:timers/promises";
+import { isDeepStrictEqual } from "node:util";
 import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
 import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime";
 import type { QaRunnerCliRegistration } from "openclaw/plugin-sdk/qa-runner-runtime";
@@ -20,6 +21,7 @@ type MatrixQaHarness = Awaited<ReturnType<typeof startMatrixQaHarness>>;
 type MatrixQaScenarioEnvironmentParams = {
  accountId: string;
  harness: MatrixQaHarness;
+  onTransportInterruptionStateChange?: (active: boolean) => void;
  observedEvents: MatrixQaObservedEvent[];
  provisioning: MatrixQaProvisionResult;
 };
@@ -42,6 +44,16 @@ type MatrixQaConfigApplyStatus = {
  hash?: string;
 };

+function resetMatrixQaScenarioObserverState(params: {
+  syncState: MatrixQaScenarioContext["syncState"];
+  syncStreams: NonNullable<MatrixQaScenarioContext["syncStreams"]>;
+}) {
+  delete params.syncState.driver;
+  delete params.syncState.observer;
+  delete params.syncStreams.driver;
+  delete params.syncStreams.observer;
+}
+
 function readMatrixConfigOverrides(
  config: Record<string, unknown>,
 ): MatrixQaConfigOverrides | undefined {
@@ -51,14 +63,23 @@ function readMatrixConfigOverrides(
    : undefined;
 }

-function resolveMatrixQaReplacePaths(overrides: MatrixQaConfigOverrides | undefined) {
-  const replacePaths = ["channels.matrix", "messages"];
+function resolveMatrixQaReplacePaths(params: {
+  accountId: string;
+  overrides: MatrixQaConfigOverrides | undefined;
+}) {
+  // Scenario topology rebuilds groupAllowFrom and may shrink it. The Gateway
+  // requires the exact array path so a parent replacement cannot bypass its guard.
+  const replacePaths = [
+    "channels.matrix",
+    `channels.matrix.accounts.${params.accountId}.groupAllowFrom`,
+    "messages",
+  ];
  // Replacing an untouched root drops config.get-omitted runtime policy and can
  // invalidate lifecycle-owned state while the Matrix account is restarting.
-  if (overrides?.agentDefaults) {
+  if (params.overrides?.agentDefaults) {
    replacePaths.push("agents.defaults");
  }
-  if (overrides?.toolProfile || overrides?.audio) {
+  if (params.overrides?.toolProfile || params.overrides?.audio) {
    replacePaths.push("tools");
  }
  return replacePaths;
@@ -212,6 +233,10 @@ export function createMatrixQaScenarioEnvironment(params: MatrixQaScenarioEnviro
      sutUserId: params.provisioning.sut.userId,
      topology: params.provisioning.topology,
    });
+    const matrixConfigChanged = !isDeepStrictEqual(
+      configSnapshot.config.channels?.matrix,
+      gatewayConfig.channels?.matrix,
+    );
    const patchStartedAt = Date.now();
    const accountStartAtBeforePatch = (
      await readMatrixAccountStatuses(input.gateway).catch(() => [])
@@ -219,7 +244,10 @@ export function createMatrixQaScenarioEnvironment(params: MatrixQaScenarioEnviro
    const patchResult = await patchGatewayConfig({
      gateway: input.gateway,
      patch: gatewayConfig as Record<string, unknown>,
-      replacePaths: resolveMatrixQaReplacePaths(configOverrides),
+      replacePaths: resolveMatrixQaReplacePaths({
+        accountId: params.accountId,
+        overrides: configOverrides,
+      }),
    });
    if (patchResult.noop !== true) {
      await input.waitForConfigRestartSettle({
@@ -238,16 +266,21 @@ export function createMatrixQaScenarioEnvironment(params: MatrixQaScenarioEnviro
      timeoutMs: input.timeoutMs,
    });
    await waitForMatrixAccountReady({
-      // Restart-required writes acknowledge before SIGUSR1 completes. Require a
-      // fresh Matrix account start so the scenario cannot race the old gateway.
+      // Config writes acknowledge persisted state before a deferred channel
+      // reload completes. Require the changed Matrix account to actually restart
+      // so a later config patch cannot supersede this scenario's live runtime.
      afterStartAt:
-        patchResult.sentinel?.payload?.stats?.requiresRestart === true
+        matrixConfigChanged && patchResult.noop !== true
          ? (accountStartAtBeforePatch ?? patchStartedAt - 1)
          : undefined,
      accountId: params.accountId,
      gateway: input.gateway,
      timeoutMs: input.timeoutMs,
    });
+    // Scenario actors must prime after each config/reload boundary. Reusing an
+    // observer across channel restarts can retain an in-flight timeline cursor
+    // and consume the next scenario's first preview before its predicate exists.
+    resetMatrixQaScenarioObserverState({ syncState, syncStreams });

    const scenarioContext = {
      baseUrl: params.harness.baseUrl,
@@ -313,12 +346,17 @@ export function createMatrixQaScenarioEnvironment(params: MatrixQaScenarioEnviro
        });
      },
      interruptTransport: async () => {
-        await params.harness.restartService();
-        await waitForMatrixAccountReady({
-          accountId: params.accountId,
-          gateway: input.gateway,
-          timeoutMs: Math.max(input.timeoutMs, 90_000),
-        });
+        params.onTransportInterruptionStateChange?.(true);
+        try {
+          await params.harness.restartService();
+          await waitForMatrixAccountReady({
+            accountId: params.accountId,
+            gateway: input.gateway,
+            timeoutMs: Math.max(input.timeoutMs, 90_000),
+          });
+        } finally {
+          params.onTransportInterruptionStateChange?.(false);
+        }
      },
      roomId: params.provisioning.roomId,
      sutAccountId: params.accountId,
--- a/extensions/qa-lab/src/live-transports/matrix/scenarios/scenario-runtime-reaction.test.ts
+++ b/extensions/qa-lab/src/live-transports/matrix/scenarios/scenario-runtime-reaction.test.ts
@@ -0,0 +1,84 @@
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { observeReactionScenario } from "./scenario-runtime-reaction.js";
+
+const baseUrl = "http://127.0.0.1:28008";
+const roomId = "!room:matrix-qa.test";
+
+function jsonResponse(body: unknown) {
+  return new Response(JSON.stringify(body), {
+    status: 200,
+    headers: { "content-type": "application/json" },
+  });
+}
+
+describe("Matrix reaction scenarios", () => {
+  afterEach(() => {
+    vi.unstubAllGlobals();
+  });
+
+  it("observes the sender echo from a scenario-owned cursor", async () => {
+    const syncSince: Array<string | null> = [];
+    const fetchMock = vi.fn<typeof fetch>(async (input) => {
+      const url = new URL(
+        typeof input === "string" ? input : input instanceof URL ? input.href : input.url,
+      );
+      if (url.pathname.endsWith("/sync")) {
+        const since = url.searchParams.get("since");
+        syncSince.push(since);
+        return jsonResponse(
+          since
+            ? {
+                next_batch: "reaction-next",
+                rooms: {
+                  join: {
+                    [roomId]: {
+                      timeline: {
+                        events: [
+                          {
+                            event_id: "$reaction",
+                            sender: "@driver:matrix-qa.test",
+                            type: "m.reaction",
+                            content: {
+                              "m.relates_to": {
+                                rel_type: "m.annotation",
+                                event_id: "$target",
+                                key: "👍",
+                              },
+                            },
+                          },
+                        ],
+                      },
+                    },
+                  },
+                },
+              }
+            : { next_batch: "reaction-start" },
+        );
+      }
+      if (url.pathname.includes("/send/m.reaction/")) {
+        return jsonResponse({ event_id: "$reaction" });
+      }
+      throw new Error(`unexpected Matrix QA request: ${url}`);
+    });
+    vi.stubGlobal("fetch", fetchMock);
+
+    const observedEvents: Parameters<typeof observeReactionScenario>[0]["observedEvents"] = [];
+    const result = await observeReactionScenario({
+      accessToken: "driver-token",
+      actorId: "driver",
+      actorUserId: "@driver:matrix-qa.test",
+      baseUrl,
+      observedEvents,
+      reactionTargetEventId: "$target",
+      roomId,
+      timeoutMs: 1_000,
+    });
+
+    expect(syncSince).toEqual([null, "reaction-start"]);
+    expect(result.event.eventId).toBe("$reaction");
+    expect(result.event.reaction).toEqual({ eventId: "$target", key: "👍" });
+    expect(result.startSince).toBe("reaction-start");
+    expect(result.since).toBe("reaction-next");
+    expect(observedEvents).toContainEqual(result.event);
+  });
+});
--- a/extensions/qa-lab/src/live-transports/matrix/scenarios/scenario-runtime-reaction.ts
+++ b/extensions/qa-lab/src/live-transports/matrix/scenarios/scenario-runtime-reaction.ts
@@ -1,13 +1,12 @@
 // QA Lab Matrix plugin module implements scenario runtime reaction behavior.
+
+import { createMatrixQaClient } from "../substrate/client.js";
 import type { MatrixQaObservedEvent } from "../substrate/events.js";
 import {
  advanceMatrixQaActorCursor,
  assertNoSutReplyWindow,
-  createMatrixQaDriverScenarioClient,
-  primeMatrixQaActorCursor,
  type MatrixQaActorId,
  type MatrixQaScenarioContext,
-  type MatrixQaSyncState,
 } from "./scenario-runtime-shared.js";
 import type { MatrixQaScenarioExecution } from "./scenario-types.js";

@@ -47,18 +46,21 @@ export async function observeReactionScenario(params: {
  reactionEmoji?: string;
  reactionTargetEventId: string;
  roomId: string;
-  syncState: MatrixQaSyncState;
-  syncStreams?: MatrixQaScenarioContext["syncStreams"];
  timeoutMs: number;
 }) {
-  const { client, startSince } = await primeMatrixQaActorCursor({
+  // A shared actor stream may observe the sender's remote echo between the send
+  // response and predicate registration. Prime a scenario-owned cursor so the
+  // reaction and any follow-up observation share one deterministic boundary.
+  const client = createMatrixQaClient({
    accessToken: params.accessToken,
-    actorId: params.actorId,
    baseUrl: params.baseUrl,
-    observedEvents: params.observedEvents,
-    syncState: params.syncState,
-    syncStreams: params.syncStreams,
  });
+  const startSince = await client.primeRoom();
+  if (!startSince) {
+    throw new Error(
+      `Matrix ${params.actorId} reaction observer did not return a next_batch cursor`,
+    );
+  }
  const reactionEmoji = params.reactionEmoji ?? "👍";
  const reactionEventId = await client.sendReaction({
    emoji: reactionEmoji,
@@ -81,6 +83,7 @@ export async function observeReactionScenario(params: {
  return {
    actorId: params.actorId,
    actorUserId: params.actorUserId,
+    client,
    event: matched.event,
    reactionEmoji,
    reactionEventId,
@@ -119,8 +122,6 @@ export async function runReactionNotificationScenario(context: MatrixQaScenarioC
    observedEvents: context.observedEvents,
    reactionTargetEventId,
    roomId: context.roomId,
-    syncState: context.syncState,
-    syncStreams: context.syncStreams,
    timeoutMs: context.timeoutMs,
  });
  return {
@@ -148,14 +149,11 @@ export async function runReactionNotAReplyScenario(context: MatrixQaScenarioCont
    observedEvents: context.observedEvents,
    reactionTargetEventId,
    roomId: context.roomId,
-    syncState: context.syncState,
-    syncStreams: context.syncStreams,
    timeoutMs: context.timeoutMs,
  });
-  const client = createMatrixQaDriverScenarioClient(context);
  const { noReplyWindowMs } = await assertNoSutReplyWindow({
    actorId: reaction.actorId,
-    client,
+    client: reaction.client,
    context,
    roomId: context.roomId,
    since: reaction.since,
@@ -196,17 +194,14 @@ export async function runReactionRedactionObservedScenario(context: MatrixQaScen
    observedEvents: context.observedEvents,
    reactionTargetEventId,
    roomId: context.roomId,
-    syncState: context.syncState,
-    syncStreams: context.syncStreams,
    timeoutMs: context.timeoutMs,
  });
-  const client = createMatrixQaDriverScenarioClient(context);
-  const redactionEventId = await client.redactEvent({
+  const redactionEventId = await reaction.client.redactEvent({
    eventId: reaction.reactionEventId,
    reason: "matrix qa reaction removal",
    roomId: context.roomId,
  });
-  const redaction = await client.waitForRoomEvent({
+  const redaction = await reaction.client.waitForRoomEvent({
    observedEvents: context.observedEvents,
    predicate: (event) =>
      event.roomId === context.roomId &&
--- a/extensions/qa-lab/src/live-transports/matrix/scenarios/scenario-runtime-room.ts
+++ b/extensions/qa-lab/src/live-transports/matrix/scenarios/scenario-runtime-room.ts
@@ -357,8 +357,6 @@ export async function runReactionThreadedScenario(context: MatrixQaScenarioConte
    observedEvents: context.observedEvents,
    reactionTargetEventId: thread.reply.eventId,
    roomId: context.roomId,
-    syncState: context.syncState,
-    syncStreams: context.syncStreams,
    timeoutMs: context.timeoutMs,
  });
  advanceMatrixQaActorCursor({
--- a/extensions/qa-lab/src/live-transports/matrix/substrate/config.test.ts
+++ b/extensions/qa-lab/src/live-transports/matrix/substrate/config.test.ts
@@ -265,10 +265,48 @@ describe("matrix qa config", () => {
    expect(account?.streaming).toEqual({
      block: { enabled: true },
      mode: "partial",
+      preview: { toolProgress: true },
    });
    expect(config.messages?.groupChat?.mentionPatterns).toEqual(["\\S"]);
  });

+  it("resets tool progress when a scalar streaming override follows an opt-out", () => {
+    const optedOut = buildMatrixQaConfig({} as OpenClawConfig, {
+      driverUserId: "@driver:matrix-qa.test",
+      homeserver: "http://127.0.0.1:28008/",
+      observerUserId: "@observer:matrix-qa.test",
+      overrides: {
+        streaming: {
+          mode: "quiet",
+          preview: { toolProgress: false },
+        },
+      },
+      sutAccessToken: "sut-token",
+      sutAccountId: "sut",
+      sutUserId: "@sut:matrix-qa.test",
+      topology,
+    });
+    const reset = buildMatrixQaConfig(optedOut, {
+      driverUserId: "@driver:matrix-qa.test",
+      homeserver: "http://127.0.0.1:28008/",
+      observerUserId: "@observer:matrix-qa.test",
+      overrides: { streaming: "quiet" },
+      sutAccessToken: "sut-token",
+      sutAccountId: "sut",
+      sutUserId: "@sut:matrix-qa.test",
+      topology,
+    });
+
+    expect(optedOut.channels?.matrix?.accounts?.sut?.streaming).toEqual({
+      mode: "quiet",
+      preview: { toolProgress: false },
+    });
+    expect(reset.channels?.matrix?.accounts?.sut?.streaming).toEqual({
+      mode: "quiet",
+      preview: { toolProgress: true },
+    });
+  });
+
  it("applies Matrix approval delivery overrides with gateway forwarding enabled", () => {
    const next = buildMatrixQaConfig({} as OpenClawConfig, {
      driverUserId: "@driver:matrix-qa.test",
--- a/extensions/qa-lab/src/live-transports/matrix/substrate/config.ts
+++ b/extensions/qa-lab/src/live-transports/matrix/substrate/config.ts
@@ -439,11 +439,10 @@ function buildMatrixQaChannelAccountConfig(params: {
  // their scalar/boolean vocabulary and normalize here before config write.
  const streamingSlots = {
    ...(params.overrides?.streaming !== undefined
-      ? { mode: resolveMatrixQaStreamingMode(params.overrides.streaming) }
-      : {}),
-    ...(isMatrixQaStreamingConfig(params.overrides?.streaming) &&
-    params.overrides.streaming.preview?.toolProgress !== undefined
-      ? { preview: { toolProgress: params.overrides.streaming.preview.toolProgress } }
+      ? {
+          mode: resolveMatrixQaStreamingMode(params.overrides.streaming),
+          preview: { toolProgress: params.snapshot.streamingPreviewToolProgress },
+        }
      : {}),
    ...(params.snapshot.chunkMode !== undefined ? { chunkMode: params.snapshot.chunkMode } : {}),
    ...(params.overrides?.blockStreaming !== undefined
--- a/extensions/qa-lab/src/live-transports/matrix/substrate/fault-proxy.ts
+++ b/extensions/qa-lab/src/live-transports/matrix/substrate/fault-proxy.ts
@@ -85,6 +85,7 @@ export type MatrixQaFaultProxyHit = {
 type MatrixQaFaultProxy = {
  baseUrl: string;
  hits(): MatrixQaFaultProxyHit[];
+  setTargetBaseUrl(targetBaseUrl: string): void;
  stop(): Promise<void>;
 };

@@ -326,7 +327,7 @@ export async function startMatrixQaFaultProxy(
    targetBaseUrl: string;
  },
 ): Promise<MatrixQaFaultProxy> {
-  const targetBaseUrl = new URL(params.targetBaseUrl);
+  let targetBaseUrl = new URL(params.targetBaseUrl);
  const maxRequestBytes = params.maxRequestBytes ?? DEFAULT_FAULT_PROXY_REQUEST_MAX_BYTES;
  const maxResponseBytes = params.maxResponseBytes ?? DEFAULT_FAULT_PROXY_RESPONSE_MAX_BYTES;
  const hits: MatrixQaFaultProxyHit[] = [];
@@ -461,6 +462,9 @@ export async function startMatrixQaFaultProxy(
  return {
    baseUrl: `http://127.0.0.1:${address.port}`,
    hits: () => [...hits],
+    setTargetBaseUrl(nextTargetBaseUrl) {
+      targetBaseUrl = new URL(nextTargetBaseUrl);
+    },
    stop: async () => {
      const closePromise = new Promise<void>((resolve, reject) => {
        server.close((error) => {
--- a/extensions/qa-lab/src/live-transports/matrix/substrate/harness.runtime.test.ts
+++ b/extensions/qa-lab/src/live-transports/matrix/substrate/harness.runtime.test.ts
@@ -43,6 +43,7 @@ async function withStartedMatrixHarness(
          buildManifest: vi.fn(),
          records: () => [],
          setScenarioId: vi.fn(),
+          setTargetBaseUrl: vi.fn(),
          stop: vi.fn(async () => {}),
        }) as unknown as MatrixQaRecordingProxy);
    const result = await startMatrixQaHarness(
@@ -152,15 +153,50 @@ describe("matrix harness runtime", () => {
    );
  });

+  it("bounds the full homeserver restart and readiness phase", async () => {
+    vi.useFakeTimers();
+    try {
+      await withStartedMatrixHarness(
+        {
+          async runCommand(_command, args) {
+            const rendered = args.join(" ");
+            if (rendered.includes("restart matrix-qa-homeserver")) {
+              return await new Promise<never>(() => {});
+            }
+            if (rendered.includes("ps --format json")) {
+              return { stdout: '[{"State":"running"}]\n', stderr: "" };
+            }
+            return { stdout: "", stderr: "" };
+          },
+          fetchImpl: vi.fn(async () => ({ ok: true })),
+          sleepImpl: vi.fn(async () => {}),
+        },
+        async ({ result }) => {
+          const restarting = result.restartService();
+          const rejection = expect(restarting).rejects.toThrow(
+            `Matrix homeserver restart timed out after ${MATRIX_QA_CLEANUP_TIMEOUT_MS}ms`,
+          );
+          await vi.advanceTimersByTimeAsync(MATRIX_QA_CLEANUP_TIMEOUT_MS);
+          await rejection;
+        },
+      );
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
  it("lets Docker atomically assign an unpinned loopback port", async () => {
    const calls: string[] = [];
+    const setTargetBaseUrl = vi.fn();
+    let portReadCount = 0;
    await withStartedMatrixHarness(
      {
        async runCommand(command, args, cwd) {
          calls.push([command, ...args, `@${cwd}`].join(" "));
          const rendered = args.join(" ");
          if (rendered.includes("port matrix-qa-homeserver 8008")) {
-            return { stdout: "127.0.0.1:49152\n", stderr: "" };
+            portReadCount += 1;
+            return { stdout: `127.0.0.1:${portReadCount === 1 ? 49_152 : 49_153}\n`, stderr: "" };
          }
          if (rendered.includes("ps --format json")) {
            return { stdout: '[{"State":"running"}]\n', stderr: "" };
@@ -169,6 +205,16 @@ describe("matrix harness runtime", () => {
        },
        fetchImpl: vi.fn(async () => ({ ok: true })),
        sleepImpl: vi.fn(async () => {}),
+        startRecordingProxyImpl: vi.fn(async ({ targetBaseUrl }) => ({
+          baseUrl: targetBaseUrl,
+          buildManifest: vi.fn(),
+          createExchangeContext: vi.fn(),
+          onExchange: vi.fn(),
+          records: () => [],
+          setScenarioId: vi.fn(),
+          setTargetBaseUrl,
+          stop: vi.fn(async () => {}),
+        })),
      },
      async ({ outputDir, result }) => {
        expect(result.homeserverPort).toBe(49152);
@@ -178,6 +224,8 @@ describe("matrix harness runtime", () => {
        );
        const compose = await readFile(result.composeFile, "utf8");
        expect(compose).toContain("      - target: 8008\n        host_ip: 127.0.0.1");
+        await result.restartService();
+        expect(setTargetBaseUrl).toHaveBeenCalledWith("http://127.0.0.1:49153/");
      },
      { dynamicPort: true },
    );
--- a/extensions/qa-lab/src/live-transports/matrix/substrate/harness.runtime.ts
+++ b/extensions/qa-lab/src/live-transports/matrix/substrate/harness.runtime.ts
@@ -90,44 +90,7 @@ export async function startMatrixQaHarness(
        ).stdout;
    const homeserverPort =
      requestedHomeserverPort || parseMatrixQaPublishedPort(publishedPortOutput);
-    await sleepImpl(1_000);
-    await waitForDockerServiceHealth(
-      MATRIX_QA_SERVICE,
-      files.composeFile,
-      repoRoot,
-      runCommand,
-      sleepImpl,
-    );
-
-    const hostBaseUrl = `http://127.0.0.1:${homeserverPort}/`;
-    let upstreamBaseUrl = hostBaseUrl;
-    const hostReachable = await isMatrixVersionsReachable(hostBaseUrl, fetchImpl);
-    if (!hostReachable) {
-      const containerBaseUrl = await resolveComposeServiceUrl(
-        MATRIX_QA_SERVICE,
-        MATRIX_QA_INTERNAL_PORT,
-        files.composeFile,
-        repoRoot,
-        runCommand,
-      );
-      upstreamBaseUrl = await waitForReachableMatrixBaseUrl({
-        composeFile: files.composeFile,
-        containerBaseUrl,
-        fetchImpl,
-        hostBaseUrl,
-        sleepImpl,
-      });
-    }
-
-    await waitForHealth(buildVersionsUrl(upstreamBaseUrl), {
-      label: "Matrix homeserver",
-      composeFile: files.composeFile,
-      fetchImpl,
-      sleepImpl,
-    });
-    const recording = await startRecordingProxyImpl({ targetBaseUrl: upstreamBaseUrl });
-
-    const waitForReady = async () => {
+    const resolveReadyUpstreamBaseUrl = async (publishedPort: number) => {
      await sleepImpl(1_000);
      await waitForDockerServiceHealth(
        MATRIX_QA_SERVICE,
@@ -136,13 +99,34 @@ export async function startMatrixQaHarness(
        runCommand,
        sleepImpl,
      );
-      await waitForHealth(buildVersionsUrl(upstreamBaseUrl), {
+      const hostBaseUrl = `http://127.0.0.1:${publishedPort}/`;
+      let readyBaseUrl = hostBaseUrl;
+      if (!(await isMatrixVersionsReachable(hostBaseUrl, fetchImpl))) {
+        const containerBaseUrl = await resolveComposeServiceUrl(
+          MATRIX_QA_SERVICE,
+          MATRIX_QA_INTERNAL_PORT,
+          files.composeFile,
+          repoRoot,
+          runCommand,
+        );
+        readyBaseUrl = await waitForReachableMatrixBaseUrl({
+          composeFile: files.composeFile,
+          containerBaseUrl,
+          fetchImpl,
+          hostBaseUrl,
+          sleepImpl,
+        });
+      }
+      await waitForHealth(buildVersionsUrl(readyBaseUrl), {
        label: "Matrix homeserver",
        composeFile: files.composeFile,
        fetchImpl,
        sleepImpl,
      });
+      return readyBaseUrl;
    };
+    let upstreamBaseUrl = await resolveReadyUpstreamBaseUrl(homeserverPort);
+    const recording = await startRecordingProxyImpl({ targetBaseUrl: upstreamBaseUrl });

    return {
      ...files,
@@ -150,12 +134,37 @@ export async function startMatrixQaHarness(
      baseUrl: recording.baseUrl,
      recording,
      async restartService() {
-        await runCommand(
-          "docker",
-          ["compose", "-f", files.composeFile, "restart", MATRIX_QA_SERVICE],
-          repoRoot,
+        await withMatrixQaHarnessTimeout(
+          "Matrix homeserver restart",
+          MATRIX_QA_CLEANUP_TIMEOUT_MS,
+          (async () => {
+            await runCommand(
+              "docker",
+              ["compose", "-f", files.composeFile, "restart", MATRIX_QA_SERVICE],
+              repoRoot,
+            );
+            const restartedPort = requestedHomeserverPort
+              ? homeserverPort
+              : parseMatrixQaPublishedPort(
+                  (
+                    await runCommand(
+                      "docker",
+                      [
+                        "compose",
+                        "-f",
+                        files.composeFile,
+                        "port",
+                        MATRIX_QA_SERVICE,
+                        String(MATRIX_QA_INTERNAL_PORT),
+                      ],
+                      repoRoot,
+                    )
+                  ).stdout,
+                );
+            upstreamBaseUrl = await resolveReadyUpstreamBaseUrl(restartedPort);
+            recording.setTargetBaseUrl(upstreamBaseUrl);
+          })(),
        );
-        await waitForReady();
      },
      stopCommand: `docker compose -f ${files.composeFile} down --remove-orphans`,
      async stop() {
@@ -178,7 +187,9 @@ export async function startMatrixQaHarness(
          throw new AggregateError(failures, "Matrix QA harness cleanup failed");
        }
      },
-      upstreamBaseUrl,
+      get upstreamBaseUrl() {
+        return upstreamBaseUrl;
+      },
    };
  } catch (error) {
    try {
--- a/extensions/qa-lab/src/live-transports/matrix/substrate/recording-proxy.ts
+++ b/extensions/qa-lab/src/live-transports/matrix/substrate/recording-proxy.ts
@@ -111,6 +111,7 @@ export type MatrixQaRecordingProxy = MatrixQaFaultProxyObserver & {
  }): MatrixQaRouteStateManifest;
  records(): MatrixQaRecordedExchange[];
  setScenarioId(scenarioId: string): void;
+  setTargetBaseUrl(targetBaseUrl: string): void;
  stop(): Promise<void>;
 };

@@ -626,6 +627,7 @@ export async function startMatrixQaRecordingProxy(params: {
    setScenarioId(nextScenarioId) {
      scenarioId = nextScenarioId;
    },
+    setTargetBaseUrl: (targetBaseUrl) => proxy.setTargetBaseUrl(targetBaseUrl),
    stop: () => proxy.stop(),
  };
 }
--- a/extensions/qa-lab/src/providers/mock-openai/mock-openai-directives.ts
+++ b/extensions/qa-lab/src/providers/mock-openai/mock-openai-directives.ts
@@ -23,16 +23,6 @@ function extractCaptures(text: string, pattern: RegExp) {
  return Array.from(text.matchAll(globalPattern), (match) => match[1]?.trim()).filter(Boolean);
 }

-export function extractLastMatchingUserText(texts: string[], pattern: RegExp) {
-  for (let index = texts.length - 1; index >= 0; index -= 1) {
-    const text = texts[index] ?? "";
-    if (pattern.test(text)) {
-      return text;
-    }
-  }
-  return "";
-}
-
 export function extractExactReplyDirective(text: string) {
  const backtickedMatch = extractLastCapture(text, /reply(?: with)? exactly\s+`([^`]+)`/i);
  if (backtickedMatch) {
--- a/extensions/qa-lab/src/providers/mock-openai/mock-openai-input.ts
+++ b/extensions/qa-lab/src/providers/mock-openai/mock-openai-input.ts
@@ -56,7 +56,7 @@ function isInternalRuntimeContextCarrierText(text: string) {
  );
 }

-function isToolOutputContinuationText(text: string) {
+function isContinuationUserText(text: string) {
  const trimmed = text.trim();
  if (!trimmed) {
    return false;
@@ -167,7 +167,7 @@ export function extractToolOutput(input: ResponsesInputItem[]) {
        .filter(Boolean);
      if (
        laterUserTexts.length > 0 &&
-        laterUserTexts.every((text) => isToolOutputContinuationText(text))
+        laterUserTexts.every((text) => isContinuationUserText(text))
      ) {
        return output;
      }
@@ -195,7 +195,7 @@ export function extractToolOutputStructuredError(input: ResponsesInputItem[]) {
        .filter(Boolean);
      if (
        laterUserTexts.length > 0 &&
-        laterUserTexts.every((text) => isToolOutputContinuationText(text))
+        laterUserTexts.every((text) => isContinuationUserText(text))
      ) {
        return functionCallOutputIsStructuredError(candidateItem);
      }
@@ -222,7 +222,7 @@ export function extractToolOutputCallId(input: ResponsesInputItem[]) {
        .filter(Boolean);
      if (
        laterUserTexts.length > 0 &&
-        laterUserTexts.every((text) => isToolOutputContinuationText(text))
+        laterUserTexts.every((text) => isContinuationUserText(text))
      ) {
        return extractFunctionCallOutputCallId(candidateItem);
      }
--- a/extensions/qa-lab/src/providers/mock-openai/mock-openai-tooling.ts
+++ b/extensions/qa-lab/src/providers/mock-openai/mock-openai-tooling.ts
@@ -40,9 +40,10 @@ export function readTargetFromPrompt(prompt: string) {
    return repoScoped;
  }

-  const loosePath = /\b[A-Za-z0-9._-]+\.(?:md|json|ts|tsx|js|mjs|cjs|txt|yaml|yml)\b/i
-    .exec(prompt)?.[0]
-    ?.trim();
+  const loosePath =
+    /\b[A-Za-z0-9_][A-Za-z0-9._@!:-]*\.(?:md|json|ts|tsx|js|mjs|cjs|txt|yaml|yml)\b/i
+      .exec(prompt)?.[0]
+      ?.trim();
  if (loosePath) {
    return loosePath;
  }
--- a/extensions/qa-lab/src/providers/mock-openai/server.test.ts
+++ b/extensions/qa-lab/src/providers/mock-openai/server.test.ts
@@ -1,6 +1,7 @@
 // Qa Lab tests cover server plugin behavior.
 import { afterEach, describe, expect, it } from "vitest";
 import { readQaMockRequestCursor } from "../shared/debug-request-cursor.js";
+import { readTargetFromPrompt } from "./mock-openai-tooling.js";
 import { startQaMockOpenAiServer } from "./server.js";

 const cleanups: Array<() => Promise<void>> = [];
@@ -1026,6 +1027,229 @@ describe("qa mock openai server", () => {
    expect(currentNormalBody).not.toContain("stale-missing-progress-target.txt");
  });

+  it.each([
+    {
+      name: "preview",
+      prompt:
+        "@openclaw:matrix-qa.test Tool progress QA check: call the read tool exactly once on `QA_KICKOFF_TASK.md` before answering. After that read completes, reply exactly `CURRENT_PREVIEW_OK`.",
+      toolName: "read",
+      expectedArgs: { path: "QA_KICKOFF_TASK.md" },
+    },
+    {
+      name: "command preview",
+      prompt:
+        "@openclaw:matrix-qa.test Tool progress QA check: call the exec tool exactly once with this exact command before answering: `printf 'matrix-command-progress-start\\n'; sleep 2`. After that exec command completes or fails, reply exactly `CURRENT_COMMAND_OK`.",
+      toolName: "exec",
+      expectedArgs: { command: "printf 'matrix-command-progress-start\\n'; sleep 2" },
+    },
+    {
+      name: "error",
+      prompt:
+        "@openclaw:matrix-qa.test Tool progress error QA check: read `missing-matrix-tool-progress-target.txt` before answering. After the read fails, reply exactly `CURRENT_ERROR_OK`.",
+      toolName: "read",
+      expectedArgs: { path: "missing-matrix-tool-progress-target.txt" },
+    },
+    {
+      name: "mention safety",
+      prompt:
+        "@openclaw:matrix-qa.test Tool progress QA check: read the missing workspace file `matrix-progress-@room-@alice:matrix-qa.test-!room:matrix-qa.test.txt` before answering. After that read fails, reply exactly `CURRENT_MENTION_OK`.",
+      toolName: "read",
+      expectedArgs: {
+        path: "matrix-progress-@room-@alice:matrix-qa.test-!room:matrix-qa.test.txt",
+      },
+    },
+  ])("routes current Matrix $name after stale streaming history", async (fixture) => {
+    const server = await startMockServer();
+    const payload = await expectResponsesJson(server, {
+      stream: false,
+      input: [
+        makeUserInput(
+          "@openclaw:matrix-qa.test Quiet streaming QA check: reply exactly `STALE_STREAMING_OK`.",
+        ),
+        makeUserInput(fixture.prompt),
+        makeUserInput("Continue with the current Matrix QA scenario."),
+      ],
+    });
+
+    const toolCall = outputToolCall(payload, fixture.toolName);
+    expect(outputToolArgsFromItem(toolCall)).toEqual(fixture.expectedArgs);
+    expect(JSON.stringify(payload)).not.toContain("STALE_STREAMING_OK");
+  });
+
+  it("does not let a prior Matrix tool result satisfy the current progress prompt", async () => {
+    const server = await startMockServer();
+    const payload = await expectResponsesJson(server, {
+      stream: false,
+      input: [
+        makeUserInput(
+          "@openclaw:matrix-qa.test Tool progress QA check: read `stale-progress-target.txt` before answering. Reply exactly `STALE_PROGRESS_OK`.",
+        ),
+        {
+          type: "function_call_output",
+          call_id: "call_mock_read_stale_progress",
+          output: "STALE_PROGRESS_OK",
+        },
+        makeUserInput(
+          "@openclaw:matrix-qa.test Tool progress QA check: read `current-progress-target.txt` before answering. Reply exactly `CURRENT_PROGRESS_OK`.",
+        ),
+      ],
+    });
+
+    const toolCall = outputToolCall(payload, "read");
+    expect(outputToolArgsFromItem(toolCall)).toEqual({ path: "current-progress-target.txt" });
+    expect(JSON.stringify(payload)).not.toContain("STALE_PROGRESS_OK");
+  });
+
+  it.each([
+    {
+      name: "quiet streaming",
+      stalePrompt:
+        "@openclaw:matrix-qa.test Quiet streaming QA check: reply exactly `STALE_QUIET_OK`.",
+    },
+    {
+      name: "tool progress",
+      stalePrompt:
+        "@openclaw:matrix-qa.test Tool progress QA check: read `stale-progress-target.txt` before answering. Reply exactly `STALE_PROGRESS_OK`.",
+    },
+  ])("does not resurrect stale Matrix $name across an ordinary turn", async (fixture) => {
+    const server = await startMockServer();
+    const payload = await expectResponsesJson(server, {
+      stream: false,
+      input: [
+        makeUserInput(fixture.stalePrompt),
+        makeUserInput("Reply exactly `CURRENT_ORDINARY_OK`."),
+      ],
+    });
+
+    expect(outputText(payload)).toBe("CURRENT_ORDINARY_OK");
+    expect(JSON.stringify(payload)).not.toContain("stale-progress-target.txt");
+  });
+
+  it("does not treat an ordinary retry request as a Matrix scenario continuation", async () => {
+    const server = await startMockServer();
+    const payload = await expectResponsesJson(server, {
+      stream: false,
+      input: [
+        makeUserInput(
+          "@openclaw:matrix-qa.test Quiet streaming QA check: reply exactly `STALE_STREAMING_OK`.",
+        ),
+        makeUserInput(
+          "Please retry the new database operation, then reply exactly `CURRENT_RETRY_OK`.",
+        ),
+      ],
+    });
+
+    expect(outputText(payload)).toBe("CURRENT_RETRY_OK");
+    expect(JSON.stringify(payload)).not.toContain("STALE_STREAMING_OK");
+  });
+
+  it("uses the current tool result marker after stale streaming history", async () => {
+    const server = await startMockServer();
+    const currentPrompt =
+      "@openclaw:matrix-qa.test Tool progress QA check: call the read tool exactly once on `QA_KICKOFF_TASK.md` before answering. The only valid final marker is inside that file.";
+    const payload = await expectResponsesJson(server, {
+      stream: false,
+      input: [
+        makeUserInput(
+          "@openclaw:matrix-qa.test Quiet streaming QA check: reply exactly `STALE_STREAMING_OK`.",
+        ),
+        makeUserInput(currentPrompt),
+        {
+          type: "function_call_output",
+          call_id: "call_mock_read_current_progress",
+          output: "Reply with only this exact marker and no other text:\nCURRENT_PROGRESS_OK",
+        },
+      ],
+    });
+
+    expect(outputText(payload)).toBe("CURRENT_PROGRESS_OK");
+  });
+
+  it("selects tool progress after quiet streaming in one user envelope", async () => {
+    const server = await startMockServer();
+    const currentPrompt =
+      "@openclaw:matrix-qa.test Tool progress QA check: call the read tool exactly once on `QA_KICKOFF_TASK.md` before answering. The only valid final marker is inside that file.";
+    const envelope = [
+      "@openclaw:matrix-qa.test Quiet streaming QA check: reply exactly `STALE_ENVELOPE_OK`.",
+      currentPrompt,
+    ].join("\n");
+    const plan = await expectResponsesJson(server, {
+      stream: false,
+      input: [makeUserInput(envelope)],
+    });
+
+    const toolCall = outputToolCall(plan, "read");
+    expect(outputToolArgsFromItem(toolCall)).toEqual({ path: "QA_KICKOFF_TASK.md" });
+    expect(JSON.stringify(plan)).not.toContain("STALE_ENVELOPE_OK");
+
+    const final = await expectResponsesJson(server, {
+      stream: false,
+      input: [
+        makeUserInput(envelope),
+        {
+          type: "function_call_output",
+          call_id: "call_mock_read_current_envelope",
+          output: "Reply with only this exact marker and no other text:\nCURRENT_ENVELOPE_OK",
+        },
+      ],
+    });
+
+    expect(outputText(final)).toBe("CURRENT_ENVELOPE_OK");
+  });
+
+  it("selects the latest tool-progress target in one user envelope", async () => {
+    const server = await startMockServer();
+    const envelope = [
+      "Tool progress QA check: read `stale-progress-target.txt` before answering. Reply exactly `STALE_PROGRESS_OK`.",
+      "Tool progress QA check: read `current-progress-target.txt` before answering. Reply exactly `CURRENT_PROGRESS_OK`.",
+    ].join("\n");
+    const payload = await expectResponsesJson(server, {
+      stream: false,
+      input: [makeUserInput(envelope)],
+    });
+
+    const toolCall = outputToolCall(payload, "read");
+    expect(outputToolArgsFromItem(toolCall)).toEqual({ path: "current-progress-target.txt" });
+    expect(JSON.stringify(payload)).not.toContain("stale-progress-target.txt");
+  });
+
+  it("selects the latest block-streaming markers and target in one user envelope", async () => {
+    const server = await startMockServer();
+    const envelope = [
+      "Block streaming QA check: first exact marker: `STALE_ONE`; read `stale-block.txt`; second exact marker: `STALE_TWO`.",
+      "Block streaming QA check: first exact marker: `CURRENT_ONE`; read `current-block.txt`; second exact marker: `CURRENT_TWO`.",
+    ].join("\n");
+    const payload = await expectResponsesJson(server, {
+      stream: false,
+      input: [makeUserInput(envelope)],
+    });
+
+    expect(outputText(payload)).toBe("CURRENT_ONE");
+    const toolCall = outputToolCall(payload, "read");
+    expect(outputToolArgsFromItem(toolCall)).toEqual({ path: "current-block.txt" });
+    expect(JSON.stringify(payload)).not.toContain("STALE_ONE");
+    expect(JSON.stringify(payload)).not.toContain("stale-block.txt");
+  });
+
+  it("rejects successful error-progress results without a prompt directive", async () => {
+    const server = await startMockServer();
+    const payload = await expectResponsesJson(server, {
+      stream: false,
+      input: [
+        makeUserInput(
+          "Tool progress error QA check: read `missing-progress-target.txt` before answering. The final marker is supplied only after the tool runs.",
+        ),
+        {
+          type: "function_call_output",
+          call_id: "call_mock_read_unexpected_success",
+          output: "Reply with only this exact marker and no other text:\nFALSE_POSITIVE_OK",
+        },
+      ],
+    });
+
+    expect(outputText(payload)).toBe("BUG-TOOL-DID-NOT-FAIL");
+  });
+
  it("prefers path-like refs over generic quoted keys in prompts", async () => {
    const server = await startMockServer();

@@ -1059,6 +1283,15 @@ describe("qa mock openai server", () => {
    expect(debugPayload.plannedToolName).toBe("read");
  });

+  it("keeps unformatted Matrix mention-shaped filenames intact", () => {
+    expect(
+      readTargetFromPrompt(
+        "Read the missing workspace file matrix-progress-@room-@alice:matrix-qa.test-!room:matrix-qa.test.txt before answering.",
+      ),
+    ).toBe("matrix-progress-@room-@alice:matrix-qa.test-!room:matrix-qa.test.txt");
+    expect(readTargetFromPrompt("Read _fixture.json before answering.")).toBe("_fixture.json");
+  });
+
  it("reads unquoted fixture paths and honors exact replies after tool output", async () => {
    const server = await startMockServer();
    const prompt =
--- a/extensions/qa-lab/src/providers/mock-openai/server.ts
+++ b/extensions/qa-lab/src/providers/mock-openai/server.ts
@@ -84,7 +84,6 @@ import {
  buildDeterministicEmbedding,
 } from "./mock-openai-contracts.js";
 import {
-  extractLastMatchingUserText,
  extractExactReplyDirective,
  extractExactMarkerDirective,
  extractWhatsAppLocationMarkerDirective,
@@ -156,6 +155,44 @@ import {
  extractSnackPreference,
 } from "./mock-openai-tooling.js";

+const QA_STREAMING_TOOL_PROGRESS_FAMILY_PROMPT_RE =
+  /(?:partial|quiet) streaming qa check|final-only marker streaming qa check|block streaming qa check|tool progress(?: error)? qa check/i;
+const QA_STREAMING_TOOL_PROGRESS_CONTINUATION_RE =
+  /^Continue with (?:the current Matrix QA scenario|the QA scenario plan and report worked, failed, and blocked items)\.$/i;
+
+function isStreamingToolProgressContinuationText(text: string) {
+  const trimmed = text.trim();
+  return (
+    QA_STREAMING_TOOL_PROGRESS_CONTINUATION_RE.test(trimmed) ||
+    trimmed.startsWith(QA_SETTLED_TOOL_TERMINAL_CONTINUATION_NEEDLE)
+  );
+}
+
+function extractLatestScenarioFamilyPrompt(texts: string[]) {
+  let envelope = "";
+  for (const text of texts.toReversed()) {
+    if (QA_STREAMING_TOOL_PROGRESS_FAMILY_PROMPT_RE.test(text)) {
+      envelope = text;
+      break;
+    }
+    if (!isStreamingToolProgressContinuationText(text)) {
+      return "";
+    }
+  }
+  if (!envelope) {
+    return "";
+  }
+  const pattern = new RegExp(
+    QA_STREAMING_TOOL_PROGRESS_FAMILY_PROMPT_RE.source,
+    `${QA_STREAMING_TOOL_PROGRESS_FAMILY_PROMPT_RE.flags}g`,
+  );
+  let latestIndex = -1;
+  for (const match of envelope.matchAll(pattern)) {
+    latestIndex = match.index;
+  }
+  return latestIndex < 0 ? "" : envelope.slice(latestIndex);
+}
+
 async function buildResponsesPayload(
  body: Record<string, unknown>,
  scenarioState: MockScenarioState,
@@ -183,6 +220,12 @@ async function buildResponsesPayload(
  const promptExactMarkerDirective = extractExactMarkerDirective(prompt);
  const allUserTexts = extractAllUserTexts(input);
  const allUserText = allUserTexts.join("\n");
+  const scenarioFamilyPrompt = extractLatestScenarioFamilyPrompt(allUserTexts) || prompt;
+  const scenarioFamilyReplyDirective =
+    extractExactReplyDirective(scenarioFamilyPrompt) ??
+    extractExactMarkerDirective(scenarioFamilyPrompt) ??
+    extractExactReplyDirective(scenarioToolOutput) ??
+    extractExactMarkerDirective(scenarioToolOutput);
  const userExactReplyDirective =
    promptExactReplyDirective ?? extractExactReplyDirective(allUserText);
  const userExactMarkerDirective =
@@ -200,13 +243,8 @@ async function buildResponsesPayload(
  const whatsAppStickerMarker = shouldUseWhatsAppStickerMarker(prompt)
    ? extractWhatsAppStickerMarkerDirective(allInputText)
    : "";
-  const blockStreamingPrompt =
-    extractLastMatchingUserText(extractAllUserTexts(input), QA_BLOCK_STREAMING_PROMPT_RE) ||
-    prompt ||
-    allInputText;
-  const blockStreamingMarkers =
-    extractBlockStreamingMarkerDirectives(blockStreamingPrompt) ??
-    extractBlockStreamingMarkerDirectives(allInputText);
+  const blockStreamingPrompt = scenarioFamilyPrompt || prompt || allInputText;
+  const blockStreamingMarkers = extractBlockStreamingMarkerDirectives(blockStreamingPrompt);
  const isGroupChat = allInputText.includes('"is_group_chat": true');
  const isBaselineUnmentionedChannelChatter = /\bno bot ping here\b/i.test(prompt);
  const hasReasoningOnlyRetryInstruction = allInputText.includes(QA_REASONING_ONLY_RETRY_NEEDLE);
@@ -229,7 +267,6 @@ async function buildResponsesPayload(
    hasDeclaredTool(body, "sessions_yield") ||
    QA_SUBAGENT_DIRECT_FALLBACK_PROMPT_RE.test(allInputText);
  const toolProgressTurn = extractLastMatchingUserTurn(input, /tool progress(?: error)? qa check/i);
-  const toolProgressPrompt = toolProgressTurn?.text ?? "";
  // Progress scenarios share full session transcripts. Scope completion to
  // the selected prompt so an older turn's tool output cannot finish this one.
  const toolProgressToolOutput = toolProgressTurn
@@ -238,11 +275,11 @@ async function buildResponsesPayload(
  const toolProgressToolJson = parseToolOutputJson(toolProgressToolOutput);
  const buildToolProgressReadEvents = () => {
    return buildToolCallEventsWithArgs("read", {
-      path: readTargetFromPrompt(toolProgressPrompt || prompt || allInputText),
+      path: readTargetFromPrompt(scenarioFamilyPrompt),
    });
  };
  const buildToolProgressExecEvents = () => {
-    const command = execCommandFromToolProgressPrompt(toolProgressPrompt || prompt || allInputText);
+    const command = execCommandFromToolProgressPrompt(scenarioFamilyPrompt);
    return command ? buildToolCallEventsWithArgs("exec", { command }) : null;
  };
  if (QA_TOOL_LOOP_GLOBAL_BREAKER_PROMPT_RE.test(allInputText)) {
@@ -621,32 +658,34 @@ async function buildResponsesPayload(
      },
    ]);
  }
-  if (QA_FINAL_ONLY_MARKER_STREAMING_PROMPT_RE.test(allInputText) && exactReplyDirective) {
+  if (
+    QA_FINAL_ONLY_MARKER_STREAMING_PROMPT_RE.test(scenarioFamilyPrompt) &&
+    scenarioFamilyReplyDirective
+  ) {
    return buildAssistantEvents([
      {
        id: "msg_mock_final_only_marker_stream",
        phase: "final_answer",
        streamDeltas: splitMockStreamingText("QA streaming preview in progress"),
-        text: exactReplyDirective,
+        text: scenarioFamilyReplyDirective,
      },
    ]);
  }
-  if (QA_STREAMING_PROMPT_RE.test(allInputText) && exactReplyDirective) {
+  if (QA_STREAMING_PROMPT_RE.test(scenarioFamilyPrompt) && scenarioFamilyReplyDirective) {
    return buildAssistantEvents([
      {
        id: "msg_mock_quiet_stream",
        phase: "final_answer",
-        streamDeltas: splitMockStreamingText(exactReplyDirective),
-        text: exactReplyDirective,
+        streamDeltas: splitMockStreamingText(scenarioFamilyReplyDirective),
+        text: scenarioFamilyReplyDirective,
      },
    ]);
  }
  const toolProgressReplyDirective =
-    extractExactReplyDirective(toolProgressPrompt) ??
-    extractExactMarkerDirective(toolProgressPrompt) ??
    extractExactReplyDirective(toolProgressToolOutput) ??
-    extractExactMarkerDirective(toolProgressToolOutput);
-  if (QA_TOOL_PROGRESS_ERROR_PROMPT_RE.test(toolProgressPrompt)) {
+    extractExactMarkerDirective(toolProgressToolOutput) ??
+    scenarioFamilyReplyDirective;
+  if (QA_TOOL_PROGRESS_ERROR_PROMPT_RE.test(scenarioFamilyPrompt)) {
    if (!toolProgressToolOutput) {
      return buildToolProgressReadEvents();
    }
@@ -658,7 +697,7 @@ async function buildResponsesPayload(
      );
    }
  }
-  if (QA_TOOL_PROGRESS_PROMPT_RE.test(toolProgressPrompt)) {
+  if (QA_TOOL_PROGRESS_PROMPT_RE.test(scenarioFamilyPrompt)) {
    if (!toolProgressToolOutput) {
      return buildToolProgressExecEvents() ?? buildToolProgressReadEvents();
    }
@@ -666,7 +705,7 @@ async function buildResponsesPayload(
      return buildAssistantEvents(toolProgressReplyDirective);
    }
  }
-  if (QA_BLOCK_STREAMING_PROMPT_RE.test(allInputText) && blockStreamingMarkers) {
+  if (QA_BLOCK_STREAMING_PROMPT_RE.test(scenarioFamilyPrompt) && blockStreamingMarkers) {
    if (!toolOutput) {
      return buildAssistantThenToolCallEvents(
        {
--- a/extensions/qa-lab/src/scenario-catalog.test.ts
+++ b/extensions/qa-lab/src/scenario-catalog.test.ts
@@ -219,13 +219,15 @@ describe("qa scenario catalog", () => {
    ).toEqual([]);
  });

-  it("uses only graceful gateway restart for Matrix replay dedupe", () => {
+  it("uses graceful restart and isolation for Matrix replay dedupe", () => {
    const scenario = requireFlowScenario(readQaScenarioById("matrix-restart-replay-dedupe"));
+    const staleSync = requireFlowScenario(readQaScenarioById("matrix-stale-sync-replay-dedupe"));

    expect(flowContainsCall(scenario.execution.flow, "env.gateway.restart")).toBe(true);
    expect(flowContainsCall(scenario.execution.flow, "env.gateway.restartAfterStateMutation")).toBe(
      false,
    );
+    expect(staleSync.execution.suiteIsolation).toBe("isolated");
  });

  it("loads scenario-declared gateway runtime options from YAML", () => {
--- a/extensions/qa-lab/src/suite-launch.runtime.test.ts
+++ b/extensions/qa-lab/src/suite-launch.runtime.test.ts
@@ -355,7 +355,7 @@ describe("qa suite runtime launcher", () => {
      expect.objectContaining({
        adapterFactories,
        channelId: "matrix",
-        outputDir: path.join(outputDir, "flow", "matrix-isolated"),
+        outputDir: path.join(outputDir, "flow", "matrix-isolated-1"),
        scenarioIds: ["matrix-restart-resume"],
      }),
    );
@@ -370,7 +370,7 @@ describe("qa suite runtime launcher", () => {
      expect.objectContaining({
        adapterFactories,
        channelId: "matrix",
-        outputDir: path.join(outputDir, "flow", "matrix-shared"),
+        outputDir: path.join(outputDir, "flow", "matrix-isolated-2"),
        scenarioIds: ["thread-isolation"],
      }),
    );
@@ -733,14 +733,15 @@ describe("qa suite runtime launcher", () => {
    const repoRoot = await makeTempRepo("qa-suite-pluggable-same-channel-concurrency-");
    const maxActive = trackMaxActiveFlowRuns();

-    const scenarioIds = [
-      "matrix-approval-channel-target-both",
+    const isolatedScenarioId = "matrix-approval-channel-target-both";
+    const sharedScenarioIds = [
      "matrix-approval-deny-reaction",
      "matrix-approval-exec-metadata-chunked",
      "matrix-approval-exec-metadata-single-event",
      "matrix-approval-plugin-metadata-single-event",
      "matrix-approval-thread-target",
    ];
+    const scenarioIds = [isolatedScenarioId, ...sharedScenarioIds];
    await runQaSuite({
      repoRoot,
      outputDir: ".artifacts/qa-e2e/pluggable-same-channel-concurrency",
@@ -759,9 +760,10 @@ describe("qa suite runtime launcher", () => {
    });

    expect(runQaFlowSuite).toHaveBeenCalledTimes(6);
-    expect(runQaFlowSuite.mock.calls.map(([params]) => params?.scenarioIds)).toEqual(
-      scenarioIds.map((scenarioId) => [scenarioId]),
-    );
+    expect(runQaFlowSuite.mock.calls.map(([params]) => params?.scenarioIds)).toEqual([
+      ...sharedScenarioIds.map((scenarioId) => [scenarioId]),
+      [isolatedScenarioId],
+    ]);
    expect(maxActive()).toBe(6);
  });

--- a/extensions/qa-lab/src/suite-planning.test.ts
+++ b/extensions/qa-lab/src/suite-planning.test.ts
@@ -5,6 +5,7 @@ import path from "node:path";
 import { describe, expect, it, vi } from "vitest";
 import { defaultQaSuiteConcurrencyForTransport } from "./qa-transport-registry.js";
 import { readQaScenarioById } from "./scenario-catalog.js";
+import { requireFlowScenario } from "./scenario-catalog.test-utils.js";
 import {
  collectQaSuiteGatewayConfigPatch,
  collectQaSuiteGatewayRuntimeOptions,
@@ -543,6 +544,51 @@ describe("qa suite planning helpers", () => {
      ),
    ).toBe(true);
    expect(scenarioRequiresIsolatedQaSuiteWorker(makeQaSuiteTestScenario("plain"))).toBe(false);
+    expect(
+      scenarioRequiresIsolatedQaSuiteWorker(readQaScenarioById("matrix-dm-thread-reply-override")),
+    ).toBe(true);
+  });
+
+  it("isolates Matrix reaction flows that require a fresh native canary", () => {
+    const scenarioIds = [
+      "matrix-reaction-notification",
+      "matrix-reaction-threaded",
+      "matrix-reaction-not-a-reply",
+      "matrix-reaction-redaction-observed",
+    ];
+
+    for (const scenarioId of scenarioIds) {
+      const scenario = requireFlowScenario(readQaScenarioById(scenarioId));
+      expect(scenario.execution.suiteIsolation, scenarioId).toBe("isolated");
+      expect(scenario.execution.isolationReason, scenarioId).toContain("fresh canary reply");
+      expect(scenarioRequiresIsolatedQaSuiteWorker(scenario), scenarioId).toBe(true);
+    }
+  });
+
+  it("isolates only positive model-driven Matrix allowBots admission flows", () => {
+    const isolatedScenarioIds = [
+      "matrix-allowbots-mentions-mentioned-room",
+      "matrix-allowbots-room-override-enables-account-off",
+      "matrix-allowbots-true-unmentioned-open-room",
+    ];
+    const sharedScenarioIds = [
+      "matrix-allowbots-default-block",
+      "matrix-allowbots-self-sender-ignored",
+      "matrix-mention-metadata-spoof-block",
+    ];
+
+    for (const scenarioId of isolatedScenarioIds) {
+      expect(
+        scenarioRequiresIsolatedQaSuiteWorker(readQaScenarioById(scenarioId)),
+        scenarioId,
+      ).toBe(true);
+    }
+    for (const scenarioId of sharedScenarioIds) {
+      expect(
+        scenarioRequiresIsolatedQaSuiteWorker(readQaScenarioById(scenarioId)),
+        scenarioId,
+      ).toBe(false);
+    }
  });

  it("isolates and collects scenario-declared transport policy", () => {
--- a/qa/scenarios/channels/dm-shared-session.yaml
+++ b/qa/scenarios/channels/dm-shared-session.yaml
@@ -19,6 +19,8 @@ scenario:
  execution:
    kind: flow
    channel: matrix
+    suiteIsolation: isolated
+    isolationReason: Requires pristine per-user DM session routing across two rooms.
    summary: Send consecutive DMs from one Matrix user in two rooms and require the shared-session notice.
    config:
      primaryConversationId: driver-dm
--- a/qa/scenarios/channels/matrix-allowbots-mentions-mentioned-room.yaml
+++ b/qa/scenarios/channels/matrix-allowbots-mentions-mentioned-room.yaml
@@ -8,6 +8,8 @@ scenario:
  execution:
    kind: flow
    channel: matrix
+    suiteIsolation: isolated
+    isolationReason: Asserts fresh model-driven configured-bot admission and cannot inherit allowBots session history from the shared Matrix worker.
    timeoutMs: 45000
    retryCount: 0
    config:
--- a/qa/scenarios/channels/matrix-allowbots-room-override-enables-account-off.yaml
+++ b/qa/scenarios/channels/matrix-allowbots-room-override-enables-account-off.yaml
@@ -8,6 +8,8 @@ scenario:
  execution:
    kind: flow
    channel: matrix
+    suiteIsolation: isolated
+    isolationReason: Asserts fresh model-driven configured-bot admission and cannot inherit allowBots session history from the shared Matrix worker.
    timeoutMs: 45000
    retryCount: 0
    config:
--- a/qa/scenarios/channels/matrix-allowbots-true-unmentioned-open-room.yaml
+++ b/qa/scenarios/channels/matrix-allowbots-true-unmentioned-open-room.yaml
@@ -8,6 +8,8 @@ scenario:
  execution:
    kind: flow
    channel: matrix
+    suiteIsolation: isolated
+    isolationReason: Asserts fresh model-driven configured-bot admission and cannot inherit allowBots session history from the shared Matrix worker.
    timeoutMs: 45000
    retryCount: 0
    config:
--- a/qa/scenarios/channels/matrix-approval-channel-target-both.yaml
+++ b/qa/scenarios/channels/matrix-approval-channel-target-both.yaml
@@ -8,6 +8,8 @@ scenario:
  execution:
    kind: flow
    channel: matrix
+    suiteIsolation: isolated
+    isolationReason: Asserts fresh channel and DM approval fan-out and cannot inherit shared Matrix approval routing state.
    timeoutMs: 90000
    retryCount: 0
    config:
--- a/qa/scenarios/channels/matrix-dm-thread-reply-override.yaml
+++ b/qa/scenarios/channels/matrix-dm-thread-reply-override.yaml
@@ -8,6 +8,8 @@ scenario:
  execution:
    kind: flow
    channel: matrix
+    suiteIsolation: isolated
+    isolationReason: Asserts fresh DM native-thread routing after session-scope scenarios and cannot inherit their DM session binding.
    timeoutMs: 45000
    retryCount: 0
    config:
--- a/qa/scenarios/channels/matrix-homeserver-restart-resume.yaml
+++ b/qa/scenarios/channels/matrix-homeserver-restart-resume.yaml
@@ -8,6 +8,8 @@ scenario:
  execution:
    kind: flow
    channel: matrix
+    suiteIsolation: isolated
+    isolationReason: Restarts the disposable homeserver process and its active Matrix sync streams.
    timeoutMs: 75000
    retryCount: 0
    config:
--- a/qa/scenarios/channels/matrix-reaction-not-a-reply.yaml
+++ b/qa/scenarios/channels/matrix-reaction-not-a-reply.yaml
@@ -8,6 +8,8 @@ scenario:
  execution:
    kind: flow
    channel: matrix
+    suiteIsolation: isolated
+    isolationReason: Posts native reaction state against a fresh canary reply and cannot inherit a shared Matrix session.
    timeoutMs: 8000
    retryCount: 0
    config:
--- a/qa/scenarios/channels/matrix-reaction-notification.yaml
+++ b/qa/scenarios/channels/matrix-reaction-notification.yaml
@@ -8,6 +8,8 @@ scenario:
  execution:
    kind: flow
    channel: matrix
+    suiteIsolation: isolated
+    isolationReason: Posts native reaction state against a fresh canary reply and cannot inherit a shared Matrix session.
    timeoutMs: 45000
    retryCount: 0
    config:
--- a/qa/scenarios/channels/matrix-reaction-redaction-observed.yaml
+++ b/qa/scenarios/channels/matrix-reaction-redaction-observed.yaml
@@ -8,6 +8,8 @@ scenario:
  execution:
    kind: flow
    channel: matrix
+    suiteIsolation: isolated
+    isolationReason: Posts native reaction and redaction state against a fresh canary reply and cannot inherit a shared Matrix session.
    timeoutMs: 45000
    retryCount: 0
    config:
--- a/qa/scenarios/channels/matrix-reaction-threaded.yaml
+++ b/qa/scenarios/channels/matrix-reaction-threaded.yaml
@@ -8,6 +8,8 @@ scenario:
  execution:
    kind: flow
    channel: matrix
+    suiteIsolation: isolated
+    isolationReason: Posts native reaction state against a fresh canary reply and cannot inherit a shared Matrix session.
    timeoutMs: 45000
    retryCount: 0
    config:
--- a/qa/scenarios/channels/matrix-stale-sync-replay-dedupe.yaml
+++ b/qa/scenarios/channels/matrix-stale-sync-replay-dedupe.yaml
@@ -10,6 +10,7 @@ scenario:
    channel: matrix
    timeoutMs: 90000
    retryCount: 0
+    suiteIsolation: isolated
    config:
      matrixTopology:
        defaultRoomKey: main
--- a/qa/scenarios/channels/thread-isolation.yaml
+++ b/qa/scenarios/channels/thread-isolation.yaml
@@ -22,6 +22,8 @@ scenario:
  execution:
    kind: flow
    channels: [qa-channel, slack, matrix]
+    suiteIsolation: isolated
+    isolationReason: Builds thread and top-level sessions in one conversation and requires a fresh session boundary.
    summary: Establish a thread, then require a fresh top-level reply without stale relation metadata.
    config:
      threadMarker: QA-THREAD-ISOLATION-THREAD-OK
--- a/qa/scenarios/channels/thread-reply-override.yaml
+++ b/qa/scenarios/channels/thread-reply-override.yaml
@@ -21,6 +21,8 @@ scenario:
    channel: matrix
    timeoutMs: 60000
    retryCount: 0
+    suiteIsolation: isolated
+    isolationReason: Changes native thread routing and must not inherit an existing room session.
    summary: Require the canonical Matrix adapter to surface the native m.thread relation through threadId.
    config:
      conversationId: main