From d98e984ec2ae892d5238ebdec3202e16d9d223be Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Wed, 29 Jul 2026 17:23:03 -0400 Subject: [PATCH] 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 --- .../matrix/adapter.runtime.test.ts | 111 +++++++++ .../live-transports/matrix/adapter.runtime.ts | 58 ++++- .../matrix/matrix-scenario-flows.test.ts | 66 +++++ .../scenarios/scenario-environment.test.ts | 87 ++++++- .../matrix/scenarios/scenario-environment.ts | 66 +++-- .../scenario-runtime-reaction.test.ts | 84 +++++++ .../scenarios/scenario-runtime-reaction.ts | 37 ++- .../matrix/scenarios/scenario-runtime-room.ts | 2 - .../matrix/substrate/config.test.ts | 38 +++ .../matrix/substrate/config.ts | 9 +- .../matrix/substrate/fault-proxy.ts | 6 +- .../matrix/substrate/harness.runtime.test.ts | 50 +++- .../matrix/substrate/harness.runtime.ts | 101 ++++---- .../matrix/substrate/recording-proxy.ts | 2 + .../mock-openai/mock-openai-directives.ts | 10 - .../mock-openai/mock-openai-input.ts | 8 +- .../mock-openai/mock-openai-tooling.ts | 7 +- .../src/providers/mock-openai/server.test.ts | 233 ++++++++++++++++++ .../src/providers/mock-openai/server.ts | 83 +++++-- .../qa-lab/src/scenario-catalog.test.ts | 4 +- .../qa-lab/src/suite-launch.runtime.test.ts | 16 +- extensions/qa-lab/src/suite-planning.test.ts | 46 ++++ qa/scenarios/channels/dm-shared-session.yaml | 2 + ...rix-allowbots-mentions-mentioned-room.yaml | 2 + ...ots-room-override-enables-account-off.yaml | 2 + ...-allowbots-true-unmentioned-open-room.yaml | 2 + .../matrix-approval-channel-target-both.yaml | 2 + .../matrix-dm-thread-reply-override.yaml | 2 + .../matrix-homeserver-restart-resume.yaml | 2 + .../channels/matrix-reaction-not-a-reply.yaml | 2 + .../matrix-reaction-notification.yaml | 2 + .../matrix-reaction-redaction-observed.yaml | 2 + .../channels/matrix-reaction-threaded.yaml | 2 + .../matrix-stale-sync-replay-dedupe.yaml | 1 + qa/scenarios/channels/thread-isolation.yaml | 2 + .../channels/thread-reply-override.yaml | 2 + 36 files changed, 1010 insertions(+), 141 deletions(-) create mode 100644 extensions/qa-lab/src/live-transports/matrix/adapter.runtime.test.ts create mode 100644 extensions/qa-lab/src/live-transports/matrix/scenarios/scenario-runtime-reaction.test.ts diff --git a/extensions/qa-lab/src/live-transports/matrix/adapter.runtime.test.ts b/extensions/qa-lab/src/live-transports/matrix/adapter.runtime.test.ts new file mode 100644 index 000000000000..a87cba9756f9 --- /dev/null +++ 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() + .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() + .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() + .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() + .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); + }); +}); diff --git a/extensions/qa-lab/src/live-transports/matrix/adapter.runtime.ts b/extensions/qa-lab/src/live-transports/matrix/adapter.runtime.ts index 8eae6276718f..42f965b0561c 100644 --- 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; + timeoutMs: number; +}): Promise { + 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(); const busMessageIds = new Map(); + 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, }); diff --git 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 index ba82482f5064..4e1c4037dc46 100644 --- 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["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"); diff --git 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 index fc521dce843c..05962937a825 100644 --- 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 }, ); diff --git a/extensions/qa-lab/src/live-transports/matrix/scenarios/scenario-environment.ts b/extensions/qa-lab/src/live-transports/matrix/scenarios/scenario-environment.ts index 25faad1ddffa..15afba776cf8 100644 --- 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>; 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; +}) { + delete params.syncState.driver; + delete params.syncState.observer; + delete params.syncStreams.driver; + delete params.syncStreams.observer; +} + function readMatrixConfigOverrides( config: Record, ): 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, - 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, diff --git 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 new file mode 100644 index 000000000000..5993f8b9439d --- /dev/null +++ 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 = []; + const fetchMock = vi.fn(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[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); + }); +}); diff --git 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 index 95ebe9ba513a..47a45351f6ec 100644 --- 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 && diff --git 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 index 05c472034ac2..76f49a0db2ea 100644 --- 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({ diff --git a/extensions/qa-lab/src/live-transports/matrix/substrate/config.test.ts b/extensions/qa-lab/src/live-transports/matrix/substrate/config.test.ts index 6b50139c2ebf..6a5a1193ad6b 100644 --- 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", diff --git a/extensions/qa-lab/src/live-transports/matrix/substrate/config.ts b/extensions/qa-lab/src/live-transports/matrix/substrate/config.ts index 0b4a8ad7c174..6e727ad04e22 100644 --- 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 diff --git a/extensions/qa-lab/src/live-transports/matrix/substrate/fault-proxy.ts b/extensions/qa-lab/src/live-transports/matrix/substrate/fault-proxy.ts index 8fe280a204f5..d06b1e49c892 100644 --- 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; }; @@ -326,7 +327,7 @@ export async function startMatrixQaFaultProxy( targetBaseUrl: string; }, ): Promise { - 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((resolve, reject) => { server.close((error) => { diff --git 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 index 18e12b940b20..ea49e9a86a3d 100644 --- 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(() => {}); + } + 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 }, ); diff --git a/extensions/qa-lab/src/live-transports/matrix/substrate/harness.runtime.ts b/extensions/qa-lab/src/live-transports/matrix/substrate/harness.runtime.ts index 1942c936c7f3..478b304e15c4 100644 --- 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 { diff --git a/extensions/qa-lab/src/live-transports/matrix/substrate/recording-proxy.ts b/extensions/qa-lab/src/live-transports/matrix/substrate/recording-proxy.ts index 16d5da70c6a0..5851baa20f08 100644 --- 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; }; @@ -626,6 +627,7 @@ export async function startMatrixQaRecordingProxy(params: { setScenarioId(nextScenarioId) { scenarioId = nextScenarioId; }, + setTargetBaseUrl: (targetBaseUrl) => proxy.setTargetBaseUrl(targetBaseUrl), stop: () => proxy.stop(), }; } diff --git a/extensions/qa-lab/src/providers/mock-openai/mock-openai-directives.ts b/extensions/qa-lab/src/providers/mock-openai/mock-openai-directives.ts index 573275a52e46..f79918524d37 100644 --- 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) { diff --git a/extensions/qa-lab/src/providers/mock-openai/mock-openai-input.ts b/extensions/qa-lab/src/providers/mock-openai/mock-openai-input.ts index 0d75a4228e3e..05fb0b84cb45 100644 --- 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); } diff --git a/extensions/qa-lab/src/providers/mock-openai/mock-openai-tooling.ts b/extensions/qa-lab/src/providers/mock-openai/mock-openai-tooling.ts index 70db2f170c63..29d616e5d30e 100644 --- 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; } diff --git a/extensions/qa-lab/src/providers/mock-openai/server.test.ts b/extensions/qa-lab/src/providers/mock-openai/server.test.ts index e1adf243860f..46660a9379b6 100644 --- 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> = []; @@ -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 = diff --git a/extensions/qa-lab/src/providers/mock-openai/server.ts b/extensions/qa-lab/src/providers/mock-openai/server.ts index 71d04137ef77..e2f0c134f83b 100644 --- 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, 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( { diff --git a/extensions/qa-lab/src/scenario-catalog.test.ts b/extensions/qa-lab/src/scenario-catalog.test.ts index b4e4a3d25c49..4e968b98c6cc 100644 --- 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", () => { diff --git a/extensions/qa-lab/src/suite-launch.runtime.test.ts b/extensions/qa-lab/src/suite-launch.runtime.test.ts index 87a06bbc5ae9..c578085e50ec 100644 --- 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); }); diff --git a/extensions/qa-lab/src/suite-planning.test.ts b/extensions/qa-lab/src/suite-planning.test.ts index 71eb288426ba..b770e552888a 100644 --- 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", () => { diff --git a/qa/scenarios/channels/dm-shared-session.yaml b/qa/scenarios/channels/dm-shared-session.yaml index bd7bffb28c60..189ad77653b0 100644 --- 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 diff --git a/qa/scenarios/channels/matrix-allowbots-mentions-mentioned-room.yaml b/qa/scenarios/channels/matrix-allowbots-mentions-mentioned-room.yaml index c16eb5bae246..904306bf2502 100644 --- 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: diff --git a/qa/scenarios/channels/matrix-allowbots-room-override-enables-account-off.yaml b/qa/scenarios/channels/matrix-allowbots-room-override-enables-account-off.yaml index 77eb97794c50..f9d2fd731efc 100644 --- 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: diff --git a/qa/scenarios/channels/matrix-allowbots-true-unmentioned-open-room.yaml b/qa/scenarios/channels/matrix-allowbots-true-unmentioned-open-room.yaml index b1b2e5656a47..52ebf11efc23 100644 --- 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: diff --git a/qa/scenarios/channels/matrix-approval-channel-target-both.yaml b/qa/scenarios/channels/matrix-approval-channel-target-both.yaml index 7a2057372a8d..9c4d43288f44 100644 --- 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: diff --git a/qa/scenarios/channels/matrix-dm-thread-reply-override.yaml b/qa/scenarios/channels/matrix-dm-thread-reply-override.yaml index 869a35e2c265..8f8d289c1991 100644 --- 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: diff --git a/qa/scenarios/channels/matrix-homeserver-restart-resume.yaml b/qa/scenarios/channels/matrix-homeserver-restart-resume.yaml index 72596b96ccf1..9b785cd7834b 100644 --- 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: diff --git a/qa/scenarios/channels/matrix-reaction-not-a-reply.yaml b/qa/scenarios/channels/matrix-reaction-not-a-reply.yaml index 4dec91844287..f7ae5802dcef 100644 --- 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: diff --git a/qa/scenarios/channels/matrix-reaction-notification.yaml b/qa/scenarios/channels/matrix-reaction-notification.yaml index 44a432bcb835..481f7e695e40 100644 --- 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: diff --git a/qa/scenarios/channels/matrix-reaction-redaction-observed.yaml b/qa/scenarios/channels/matrix-reaction-redaction-observed.yaml index f0b6078eaa70..c495b400bc14 100644 --- 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: diff --git a/qa/scenarios/channels/matrix-reaction-threaded.yaml b/qa/scenarios/channels/matrix-reaction-threaded.yaml index 0570a89aa358..9b75048d1e85 100644 --- 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: diff --git a/qa/scenarios/channels/matrix-stale-sync-replay-dedupe.yaml b/qa/scenarios/channels/matrix-stale-sync-replay-dedupe.yaml index bd76ac56fcda..94ddccbd2344 100644 --- 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 diff --git a/qa/scenarios/channels/thread-isolation.yaml b/qa/scenarios/channels/thread-isolation.yaml index e2800083ad1e..1bf11728fa26 100644 --- 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 diff --git a/qa/scenarios/channels/thread-reply-override.yaml b/qa/scenarios/channels/thread-reply-override.yaml index 31004288cdfb..ec035c10c418 100644 --- 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