diff --git a/extensions/qa-lab/src/crabline-transport.test.ts b/extensions/qa-lab/src/crabline-transport.test.ts index a2a94eb7cbb..7e1bdfd1652 100644 --- a/extensions/qa-lab/src/crabline-transport.test.ts +++ b/extensions/qa-lab/src/crabline-transport.test.ts @@ -73,6 +73,71 @@ describe("crabline transport", () => { }); }); + it("configures distinct Telegram actors for canonical sender allowlist flows", async () => { + await withTempDir("qa-crabline-transport-", async (outputDir) => { + const transport = await createQaCrablineTransportAdapter({ + outputDir, + transportPolicy: { + requireGroupMention: true, + senderAllowlist: ["driver"], + }, + selection: createSelection(), + state: createQaBusState(), + }); + + try { + expect(transport.createGatewayConfig({ baseUrl: "http://127.0.0.1:1" })).toMatchObject({ + channels: { + telegram: { + allowFrom: ["100001"], + groupAllowFrom: ["100001"], + groupPolicy: "allowlist", + }, + }, + }); + await transport.state.addInboundMessage({ + conversation: { id: "qa-routing-ordering", kind: "group" }, + senderId: "observer", + text: "observer", + }); + await transport.state.addInboundMessage({ + conversation: { id: "qa-routing-ordering", kind: "group" }, + senderId: "driver", + text: "driver", + }); + + const manifest = JSON.parse( + await fs.readFile(path.join(outputDir, OPENCLAW_CRABLINE_MANIFEST_PATH), "utf8"), + ) as { + botToken: string; + endpoints: { apiRoot: string }; + }; + const response = await fetch( + `${manifest.endpoints.apiRoot}/bot${manifest.botToken}/getUpdates`, + ); + const payload = (await response.json()) as { + result?: Array<{ message?: { from?: { id?: number }; text?: string } }>; + }; + expect(payload.result?.map((update) => update.message?.from?.id)).toEqual([100002, 100001]); + } finally { + await transport.cleanup?.(); + } + }); + }); + + it("rejects canonical sender-policy flows on non-Telegram Crabline bridges", async () => { + await withTempDir("qa-crabline-transport-", async (outputDir) => { + await expect( + createQaCrablineTransportAdapter({ + outputDir, + transportPolicy: { senderAllowlist: ["driver"] }, + selection: createSelection("matrix"), + state: createQaBusState(), + }), + ).rejects.toThrow("Crabline matrix does not support the requested group transport policy"); + }); + }); + it("injects Telegram native commands through the shared transport adapter", async () => { await withTempDir("qa-crabline-transport-", async (outputDir) => { const transport = await createQaCrablineTransportAdapter({ diff --git a/extensions/qa-lab/src/crabline-transport.ts b/extensions/qa-lab/src/crabline-transport.ts index 98fa15d8ff8..daa4033d765 100644 --- a/extensions/qa-lab/src/crabline-transport.ts +++ b/extensions/qa-lab/src/crabline-transport.ts @@ -30,6 +30,7 @@ import type { QaTransportNativeCommandInput, QaTransportOutboundEvent, QaTransportOutboundSequenceMatch, + QaTransportPolicy, QaTransportReportParams, QaTransportState, } from "./qa-transport.js"; @@ -40,6 +41,8 @@ import type { } from "./runtime-api.js"; const CRABLINE_TRANSPORT_ID = "crabline"; +const TELEGRAM_QA_DRIVER_ID = "100001"; +const TELEGRAM_QA_OBSERVER_ID = "100002"; type QaCrablineTransportState = QaTransportState & { cleanup: () => Promise; @@ -50,6 +53,14 @@ type QaCrablineTransportState = QaTransportState & { const TELEGRAM_LIFECYCLE_METHOD_RE = /\/(sendMessage|editMessageText|deleteMessage)$/u; +function resolveTelegramQaSenderId(senderId: string) { + return senderId === "driver" + ? TELEGRAM_QA_DRIVER_ID + : senderId === "observer" + ? TELEGRAM_QA_OBSERVER_ID + : senderId; +} + function readTelegramLifecycleEvent(params: { cursor: number; event: unknown; @@ -275,7 +286,16 @@ function createCrablineState(params: { } }, async addInboundMessage(input: QaBusInboundMessageInput) { - const providerInbound = params.adapter.createInbound({ input }); + const providerSenderId = + params.adapter.channel === "telegram" + ? resolveTelegramQaSenderId(input.senderId) + : input.senderId; + const providerInbound = params.adapter.createInbound({ + input: { + ...input, + senderId: providerSenderId, + }, + }); targetByProviderTarget.set(providerInbound.providerTargetKey, providerInbound.qaTarget); const providerMessageId = await postCrablineInbound({ adapter: params.adapter, @@ -307,6 +327,7 @@ function createCrablineState(params: { class QaCrablineTransport extends QaStateBackedTransportAdapter { readonly #adapter: StartedOpenClawCrablineAdapter; readonly #selection: OpenClawCrablineChannelDriverSelection; + readonly #transportPolicy?: QaTransportPolicy; readonly #state: QaCrablineTransportState; readonly sendNativeCommand?: (input: QaTransportNativeCommandInput) => Promise; readonly waitForOutboundSequence?: (input: QaTransportOutboundSequenceMatch) => Promise<{ @@ -316,6 +337,7 @@ class QaCrablineTransport extends QaStateBackedTransportAdapter { constructor(params: { adapter: StartedOpenClawCrablineAdapter; + transportPolicy?: QaTransportPolicy; selection: OpenClawCrablineChannelDriverSelection; state: QaCrablineTransportState; }) { @@ -328,6 +350,7 @@ class QaCrablineTransport extends QaStateBackedTransportAdapter { }); this.#adapter = params.adapter; this.#selection = params.selection; + this.#transportPolicy = params.transportPolicy; this.#state = params.state; if (params.selection.channel === "telegram") { this.sendNativeCommand = async (input) => { @@ -346,8 +369,39 @@ class QaCrablineTransport extends QaStateBackedTransportAdapter { } } - createGatewayConfig = (params: { baseUrl: string }): QaTransportGatewayConfig => - this.#adapter.createGatewayConfig(params) as QaTransportGatewayConfig; + createGatewayConfig = (params: { baseUrl: string }): QaTransportGatewayConfig => { + const config = this.#adapter.createGatewayConfig(params) as OpenClawConfig; + if (this.#selection.channel !== "telegram") { + return config as QaTransportGatewayConfig; + } + const senderAllowlist = this.#transportPolicy?.senderAllowlist?.map(resolveTelegramQaSenderId); + if (!this.#transportPolicy?.requireGroupMention && !senderAllowlist) { + return config as QaTransportGatewayConfig; + } + return { + ...config, + channels: { + ...config.channels, + telegram: { + ...config.channels?.telegram, + ...(senderAllowlist + ? { + allowFrom: [...senderAllowlist], + groupAllowFrom: [...senderAllowlist], + groupPolicy: "allowlist" as const, + } + : {}), + groups: { + ...config.channels?.telegram?.groups, + "*": { + ...config.channels?.telegram?.groups?.["*"], + ...(this.#transportPolicy?.requireGroupMention ? { requireMention: true } : {}), + }, + }, + }, + }, + } as QaTransportGatewayConfig; + }; waitReady = (params: { gateway: QaTransportGatewayClient; @@ -389,9 +443,18 @@ class QaCrablineTransport extends QaStateBackedTransportAdapter { export async function createQaCrablineTransportAdapter(params: { outputDir: string; + transportPolicy?: QaTransportPolicy; selection: OpenClawCrablineChannelDriverSelection; state?: QaBusState; }) { + const requiresTelegramPolicy = + params.transportPolicy?.requireGroupMention === true || + params.transportPolicy?.senderAllowlist !== undefined; + if (params.selection.channel !== "telegram" && requiresTelegramPolicy) { + throw new Error( + `Crabline ${params.selection.channel} does not support the requested group transport policy; use the Crabline Telegram bridge or a live channel adapter`, + ); + } const recorderPath = path.join( params.outputDir, "artifacts", @@ -417,5 +480,10 @@ export async function createQaCrablineTransportAdapter(params: { state: params.state ?? createQaBusState(), }); observeEvent = state.observeEvent; - return new QaCrablineTransport({ adapter, selection: params.selection, state }); + return new QaCrablineTransport({ + adapter, + transportPolicy: params.transportPolicy, + selection: params.selection, + state, + }); } diff --git a/extensions/qa-lab/src/live-transports/live-transport-adapters.test.ts b/extensions/qa-lab/src/live-transports/live-transport-adapters.test.ts index d8aba85ad44..ebb9c972860 100644 --- a/extensions/qa-lab/src/live-transports/live-transport-adapters.test.ts +++ b/extensions/qa-lab/src/live-transports/live-transport-adapters.test.ts @@ -32,11 +32,25 @@ describe("live transport adapter factories", () => { it("assigns shared thread scenarios to Slack", () => { expect(slackQaAdapterFactory.scenarioIds).toEqual([ "channel-chat-baseline", + "channel-canary", + "channel-mention-gating", + "channel-top-level-reply-shape", "thread-follow-up", "thread-isolation", ]); }); + it("keeps WhatsApp routing flows available without making them DM-safe CLI defaults", () => { + expect(whatsappQaAdapterFactory.scenarioIds).toEqual([ + "dm-chat-baseline", + "channel-canary", + "channel-dm-group-routing", + "channel-mention-gating", + "channel-top-level-reply-shape", + "whatsapp-help-command", + ]); + }); + it.each([ ["telegram", createTelegram], ["slack", createSlack], diff --git a/extensions/qa-lab/src/live-transports/shared/canonical-scenarios.test.ts b/extensions/qa-lab/src/live-transports/shared/canonical-scenarios.test.ts index 5c8994a91b5..4a57abc1759 100644 --- a/extensions/qa-lab/src/live-transports/shared/canonical-scenarios.test.ts +++ b/extensions/qa-lab/src/live-transports/shared/canonical-scenarios.test.ts @@ -22,7 +22,7 @@ import { import { loadNonYamlScenarioRefs } from "./live-transport-scenarios.js"; describe("canonical live-transport scenarios", () => { - it("loads every migrated command and session-context scenario from YAML", () => { + it("loads every migrated routing, command, and session-context scenario from YAML", () => { const telegram = listCanonicalScenarios({ ids: TELEGRAM_CANONICAL_SCENARIO_IDS, defaultIds: TELEGRAM_DEFAULT_CANONICAL_SCENARIO_IDS, diff --git a/extensions/qa-lab/src/live-transports/shared/canonical-scenarios.ts b/extensions/qa-lab/src/live-transports/shared/canonical-scenarios.ts index dbb29cef796..319e4024bfa 100644 --- a/extensions/qa-lab/src/live-transports/shared/canonical-scenarios.ts +++ b/extensions/qa-lab/src/live-transports/shared/canonical-scenarios.ts @@ -6,6 +6,8 @@ import { runQaFlowSuiteFromRuntime } from "../../suite-launch.runtime.js"; import type { LiveTransportQaCommandOptions } from "./live-transport-cli.js"; export const TELEGRAM_CANONICAL_SCENARIO_IDS = [ + "channel-canary", + "channel-mention-gating", "telegram-help-command", "telegram-commands-command", "telegram-tools-compact-command", @@ -19,6 +21,8 @@ export const TELEGRAM_CANONICAL_SCENARIO_IDS = [ ] as const; export const TELEGRAM_DEFAULT_CANONICAL_SCENARIO_IDS = [ + "channel-canary", + "channel-mention-gating", "telegram-help-command", "telegram-commands-command", "telegram-tools-compact-command", @@ -28,7 +32,15 @@ export const TELEGRAM_DEFAULT_CANONICAL_SCENARIO_IDS = [ "telegram-context-command", ] as const; +export const WHATSAPP_ROUTING_CANONICAL_SCENARIO_IDS = [ + "channel-canary", + "channel-dm-group-routing", + "channel-mention-gating", + "channel-top-level-reply-shape", +] as const; + export const WHATSAPP_CANONICAL_SCENARIO_IDS = [ + ...WHATSAPP_ROUTING_CANONICAL_SCENARIO_IDS, "whatsapp-help-command", "whatsapp-status-command", "whatsapp-commands-command", diff --git a/extensions/qa-lab/src/live-transports/slack/cli.ts b/extensions/qa-lab/src/live-transports/slack/cli.ts index b85ed41fbda..f84663296e9 100644 --- a/extensions/qa-lab/src/live-transports/slack/cli.ts +++ b/extensions/qa-lab/src/live-transports/slack/cli.ts @@ -23,7 +23,14 @@ async function runQaSlack(opts: LiveTransportQaCommandOptions) { export const slackQaAdapterFactory: NonNullable = { id: "slack", - scenarioIds: ["channel-chat-baseline", "thread-follow-up", "thread-isolation"], + scenarioIds: [ + "channel-chat-baseline", + "channel-canary", + "channel-mention-gating", + "channel-top-level-reply-shape", + "thread-follow-up", + "thread-isolation", + ], matches: ({ channelId, driver }) => driver === "live" && channelId === "slack", async create(context) { return await (await loadSlackQaAdapterRuntime()).createSlackQaTransportAdapter(context); diff --git a/extensions/qa-lab/src/live-transports/whatsapp/adapter.runtime.ts b/extensions/qa-lab/src/live-transports/whatsapp/adapter.runtime.ts index 7088f5db6ea..6539423eefc 100644 --- a/extensions/qa-lab/src/live-transports/whatsapp/adapter.runtime.ts +++ b/extensions/qa-lab/src/live-transports/whatsapp/adapter.runtime.ts @@ -3,6 +3,7 @@ import fs from "node:fs/promises"; import path from "node:path"; import { startWhatsAppQaDriverSession } from "@openclaw/whatsapp/api.js"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; +import { buildQaTarget } from "openclaw/plugin-sdk/qa-channel"; import type { QaRunnerCliRegistration } from "openclaw/plugin-sdk/qa-runner-runtime"; import { resolvePreferredOpenClawTmpDir } from "openclaw/plugin-sdk/temp-path"; import { @@ -62,16 +63,15 @@ export async function createWhatsAppQaTransportAdapter( throw error; } const accountId = options.sutAccountId?.trim() || "sut"; - const directTargets = whatsappLive.resolveWhatsAppQaMessageTargets({ + const dmTargets = whatsappLive.resolveWhatsAppQaMessageTargets({ driverPhoneE164: runtimeEnv.driverPhoneE164, scenarioTarget: "dm", sutPhoneE164: runtimeEnv.sutPhoneE164, }); - let targets = directTargets; let observedCount = driver.getObservedMessages().length; let stopped = false; let pollingError: Error | undefined; - let logicalConversationId = targets.gatewayTarget; + let logicalConversationId = dmTargets.gatewayTarget; let logicalConversationKind: "direct" | "group" = "direct"; const nativeMessageIds = new Map(); const busMessageIds = new Map(); @@ -88,7 +88,10 @@ export async function createWhatsAppQaTransportAdapter( } await context.messages.addOutboundMessage({ accountId, - to: `${logicalConversationKind === "direct" ? "dm" : "group"}:${logicalConversationId}`, + to: buildQaTarget({ + chatType: logicalConversationKind, + conversationId: logicalConversationId, + }), senderId: message.fromPhoneE164, text: message.text, timestamp: Date.parse(message.observedAt), @@ -122,11 +125,11 @@ export async function createWhatsAppQaTransportAdapter( async sendInbound(input) { heartbeat.throwIfFailed(); logicalConversationId = input.conversation.id; - logicalConversationKind = input.conversation.kind === "group" ? "group" : "direct"; - targets = whatsappLive.resolveWhatsAppQaMessageTargets({ + logicalConversationKind = input.conversation.kind === "direct" ? "direct" : "group"; + const targets = whatsappLive.resolveWhatsAppQaMessageTargets({ driverPhoneE164: runtimeEnv.driverPhoneE164, groupJid: runtimeEnv.groupJid, - scenarioTarget: logicalConversationKind === "group" ? "group" : "dm", + scenarioTarget: logicalConversationKind === "direct" ? "dm" : "group", sutPhoneE164: runtimeEnv.sutPhoneE164, }); const quotedMessageId = input.replyToId ? nativeMessageIds.get(input.replyToId) : undefined; @@ -155,8 +158,7 @@ export async function createWhatsAppQaTransportAdapter( return message; }, resetTransport: () => { - targets = directTargets; - logicalConversationId = directTargets.gatewayTarget; + logicalConversationId = dmTargets.gatewayTarget; logicalConversationKind = "direct"; nativeMessageIds.clear(); busMessageIds.clear(); @@ -167,15 +169,16 @@ export async function createWhatsAppQaTransportAdapter( authDir: sutAuthDir, dmPolicy: "allowlist", groupJid: runtimeEnv.groupJid, + overrides: options.transportPolicy?.topLevelReplies ? { replyToMode: "off" } : undefined, sutAccountId: accountId, }), waitReady: async ({ gateway }) => await whatsappLive.waitForWhatsAppChannelStable(gateway as never, accountId), buildAgentDelivery: () => ({ channel: "whatsapp", - to: targets.gatewayTarget, + to: dmTargets.gatewayTarget, replyChannel: "whatsapp", - replyTo: targets.gatewayTarget, + replyTo: dmTargets.gatewayTarget, }), async handleAction() { throw new Error("WhatsApp live QA adapter does not implement transport actions"); diff --git a/extensions/qa-lab/src/live-transports/whatsapp/cli.ts b/extensions/qa-lab/src/live-transports/whatsapp/cli.ts index 1c2fd03b1f6..7293369804f 100644 --- a/extensions/qa-lab/src/live-transports/whatsapp/cli.ts +++ b/extensions/qa-lab/src/live-transports/whatsapp/cli.ts @@ -1,4 +1,7 @@ -import { WHATSAPP_LIVE_DEFAULT_CANONICAL_SCENARIO_IDS } from "../shared/canonical-scenarios.js"; +import { + WHATSAPP_LIVE_DEFAULT_CANONICAL_SCENARIO_IDS, + WHATSAPP_ROUTING_CANONICAL_SCENARIO_IDS, +} from "../shared/canonical-scenarios.js"; // Qa Lab plugin module implements cli behavior. import { createLazyCliRuntimeLoader, @@ -25,7 +28,11 @@ export const whatsappQaAdapterFactory: NonNullable< LiveTransportQaCliRegistration["adapterFactory"] > = { id: "whatsapp", - scenarioIds: ["dm-chat-baseline", ...WHATSAPP_LIVE_DEFAULT_CANONICAL_SCENARIO_IDS], + scenarioIds: [ + "dm-chat-baseline", + ...WHATSAPP_ROUTING_CANONICAL_SCENARIO_IDS, + ...WHATSAPP_LIVE_DEFAULT_CANONICAL_SCENARIO_IDS, + ], matches: ({ channelId, driver }) => driver === "live" && channelId === "whatsapp", async create(context) { return await (await loadWhatsAppQaAdapterRuntime()).createWhatsAppQaTransportAdapter(context); diff --git a/extensions/qa-lab/src/qa-channel-transport.test.ts b/extensions/qa-lab/src/qa-channel-transport.test.ts index 974c51fbf4a..6e9c96eaf35 100644 --- a/extensions/qa-lab/src/qa-channel-transport.test.ts +++ b/extensions/qa-lab/src/qa-channel-transport.test.ts @@ -32,6 +32,24 @@ describe("qa channel transport", () => { }); }); + it("maps declared transport policy without inspecting scenario ids", () => { + const transport = createQaChannelTransport(createQaBusState(), { + requireGroupMention: true, + senderAllowlist: ["driver"], + }); + + expect(transport.createGatewayConfig({ baseUrl: "http://127.0.0.1:43123" })).toMatchObject({ + channels: { + "qa-channel": { + allowFrom: ["driver"], + groupAllowFrom: ["driver"], + groupPolicy: "allowlist", + groups: { "*": { requireMention: true } }, + }, + }, + }); + }); + it("builds agent delivery params for qa-channel replies", () => { const transport = createQaChannelTransport(createQaBusState()); diff --git a/extensions/qa-lab/src/qa-channel-transport.ts b/extensions/qa-lab/src/qa-channel-transport.ts index 292eefc084b..389321abe76 100644 --- a/extensions/qa-lab/src/qa-channel-transport.ts +++ b/extensions/qa-lab/src/qa-channel-transport.ts @@ -15,6 +15,7 @@ import type { QaTransportGatewayClient, QaTransportNativeCommandInput, QaTransportOutboundSequenceMatch, + QaTransportPolicy, QaTransportReportParams, } from "./qa-transport.js"; import { qaChannelPlugin } from "./runtime-api.js"; @@ -83,7 +84,9 @@ async function waitForQaChannelReady(params: { export function createQaChannelGatewayConfig(params: { baseUrl: string; + transportPolicy?: QaTransportPolicy; }): QaTransportGatewayConfig { + const senderAllowlist = params.transportPolicy?.senderAllowlist; return { channels: { [QA_CHANNEL_ID]: { @@ -91,7 +94,22 @@ export function createQaChannelGatewayConfig(params: { baseUrl: params.baseUrl, botUserId: "openclaw", botDisplayName: "OpenClaw QA", - allowFrom: ["*"], + allowFrom: senderAllowlist ? [...senderAllowlist] : ["*"], + ...(senderAllowlist + ? { + groupPolicy: "allowlist" as const, + groupAllowFrom: [...senderAllowlist], + } + : {}), + ...(params.transportPolicy?.requireGroupMention + ? { + groups: { + "*": { + requireMention: true, + }, + }, + } + : {}), pollTimeoutMs: 250, }, }, @@ -134,7 +152,9 @@ async function handleQaChannelAction(params: { } class QaChannelTransport extends QaStateBackedTransportAdapter { - constructor(state: QaBusState) { + readonly #transportPolicy?: QaTransportPolicy; + + constructor(state: QaBusState, transportPolicy?: QaTransportPolicy) { super({ id: QA_CHANNEL_ID, label: "qa-channel + qa-lab bus", @@ -143,9 +163,11 @@ class QaChannelTransport extends QaStateBackedTransportAdapter { supportedActions: ["delete", "edit", "react", "thread-create"], state, }); + this.#transportPolicy = transportPolicy; } - createGatewayConfig = createQaChannelGatewayConfig; + createGatewayConfig = ({ baseUrl }: { baseUrl: string }) => + createQaChannelGatewayConfig({ baseUrl, transportPolicy: this.#transportPolicy }); waitReady = waitForQaChannelReady; buildAgentDelivery = ({ target }: { target: string }) => ({ channel: QA_CHANNEL_ID, @@ -170,6 +192,9 @@ class QaChannelTransport extends QaStateBackedTransportAdapter { createReportNotes = createQaChannelReportNotes; } -export function createQaChannelTransport(state: QaBusState) { - return new QaChannelTransport(state); +export function createQaChannelTransport( + state: QaBusState, + transportPolicy?: QaTransportPolicy, +) { + return new QaChannelTransport(state, transportPolicy); } diff --git a/extensions/qa-lab/src/qa-transport-registry.ts b/extensions/qa-lab/src/qa-transport-registry.ts index df8dc891c4e..df1193bc104 100644 --- a/extensions/qa-lab/src/qa-transport-registry.ts +++ b/extensions/qa-lab/src/qa-transport-registry.ts @@ -40,7 +40,7 @@ async function createBuiltInQaTransport( context: QaTransportFactoryContext, ): Promise { if (context.driver === "qa-channel" && context.channelId === "qa-channel") { - return createQaChannelTransport(context.state); + return createQaChannelTransport(context.state, context.adapterOptions?.transportPolicy); } if (context.driver === "crabline") { const { resolveOpenClawCrablineChannelDriverSelection } = await import("@openclaw/crabline"); @@ -48,6 +48,7 @@ async function createBuiltInQaTransport( const { createQaCrablineTransportAdapter } = await import("./crabline-transport.js"); return await createQaCrablineTransportAdapter({ outputDir: context.outputDir, + transportPolicy: context.adapterOptions?.transportPolicy, selection, state: context.state, }); @@ -93,9 +94,12 @@ export function createQaTransportAdapterFactoryRegistry( } } catch (error) { const message = error instanceof Error ? error.message : String(error); - throw new Error(`failed to create QA transport ${context.driver}:${context.channelId}: ${message}`, { - cause: error, - }); + throw new Error( + `failed to create QA transport ${context.driver}:${context.channelId}: ${message}`, + { + cause: error, + }, + ); } return { adapter, diff --git a/extensions/qa-lab/src/qa-transport.ts b/extensions/qa-lab/src/qa-transport.ts index 08fa953ba4a..0136b66561d 100644 --- a/extensions/qa-lab/src/qa-transport.ts +++ b/extensions/qa-lab/src/qa-transport.ts @@ -40,6 +40,10 @@ export type QaTransportReportParams = { export type QaTransportGatewayConfig = Pick; +export type QaTransportPolicy = NonNullable< + Parameters["create"]>[0]["adapterOptions"] +>["transportPolicy"]; + export type QaTransportState = { reset: () => void | Promise; getSnapshot: () => QaBusStateSnapshot; diff --git a/extensions/qa-lab/src/scenario-catalog.ts b/extensions/qa-lab/src/scenario-catalog.ts index c352a43d7bd..b20dc2488ec 100644 --- a/extensions/qa-lab/src/scenario-catalog.ts +++ b/extensions/qa-lab/src/scenario-catalog.ts @@ -66,12 +66,19 @@ const qaScenarioChannelSchema = z message: "scenario execution channel ids must use lowercase dotted or dashed tokens", }); +const qaScenarioTransportPolicySchema = z.object({ + requireGroupMention: z.literal(true).optional(), + senderAllowlist: z.array(z.string().trim().min(1)).min(1).optional(), + topLevelReplies: z.literal(true).optional(), +}); + const qaFlowScenarioExecutionSchema = z.object({ kind: z.literal("flow").default("flow"), summary: z.string().trim().min(1).optional(), channel: qaScenarioChannelSchema.optional(), suiteIsolation: z.literal("isolated").optional(), isolationReason: z.string().trim().min(1).optional(), + transportPolicy: qaScenarioTransportPolicySchema.optional(), config: qaScenarioConfigSchema.optional(), }); diff --git a/extensions/qa-lab/src/suite-planning.test.ts b/extensions/qa-lab/src/suite-planning.test.ts index 18a867b43e6..0f0427c2913 100644 --- a/extensions/qa-lab/src/suite-planning.test.ts +++ b/extensions/qa-lab/src/suite-planning.test.ts @@ -9,6 +9,7 @@ import { collectQaSuiteGatewayConfigPatch, collectQaSuiteGatewayRuntimeOptions, collectQaSuitePluginIds, + collectQaSuiteTransportPolicy, mapQaSuiteWithConcurrency, normalizeQaSuiteConcurrency, resolveQaSuiteScenarioChannel, @@ -320,6 +321,21 @@ describe("qa suite planning helpers", () => { expect(scenarioRequiresIsolatedQaSuiteWorker(makeQaSuiteTestScenario("plain"))).toBe(false); }); + it("isolates and collects scenario-declared transport policy", () => { + const scenario = makeQaSuiteTestScenario("sender-policy", { + transportPolicy: { + requireGroupMention: true, + senderAllowlist: ["driver"], + }, + }); + + expect(scenarioRequiresIsolatedQaSuiteWorker(scenario)).toBe(true); + expect(collectQaSuiteTransportPolicy([scenario])).toEqual({ + requireGroupMention: true, + senderAllowlist: ["driver"], + }); + }); + it("collects unique scenario-declared bundled plugins in encounter order", () => { const scenarios = [ makeQaSuiteTestScenario("generic", { plugins: ["active-memory", "memory-wiki"] }), @@ -468,6 +484,20 @@ describe("qa suite planning helpers", () => { ).toBe(false); }); + it("isolates serial runs when transport policy would leak into another scenario", () => { + expect( + shouldUseIsolatedQaSuiteScenarioWorkers({ + scenarios: [ + makeQaSuiteTestScenario("dm-baseline"), + makeQaSuiteTestScenario("sender-policy", { + transportPolicy: { senderAllowlist: ["driver"] }, + }), + ], + concurrency: 1, + }), + ).toBe(true); + }); + it("keeps concurrent runs on isolated workers", () => { expect( shouldUseIsolatedQaSuiteScenarioWorkers({ diff --git a/extensions/qa-lab/src/suite-planning.ts b/extensions/qa-lab/src/suite-planning.ts index d72605af80a..915f14b6eec 100644 --- a/extensions/qa-lab/src/suite-planning.ts +++ b/extensions/qa-lab/src/suite-planning.ts @@ -263,6 +263,39 @@ function collectQaSuiteGatewayRuntimeOptions( : undefined; } +function collectQaSuiteTransportPolicy( + scenarios: ReturnType["scenarios"], +) { + let requireGroupMention = false; + let topLevelReplies = false; + let senderAllowlist: readonly string[] | undefined; + for (const scenario of scenarios) { + if (scenario.execution.kind !== "flow") { + continue; + } + const policy = scenario.execution.transportPolicy; + requireGroupMention ||= policy?.requireGroupMention === true; + topLevelReplies ||= policy?.topLevelReplies === true; + if (!policy?.senderAllowlist) { + continue; + } + if ( + senderAllowlist && + JSON.stringify(senderAllowlist) !== JSON.stringify(policy.senderAllowlist) + ) { + throw new Error("Selected QA scenarios require conflicting transport sender allowlists."); + } + senderAllowlist = policy.senderAllowlist; + } + return requireGroupMention || topLevelReplies || senderAllowlist + ? { + ...(requireGroupMention ? { requireGroupMention: true as const } : {}), + ...(senderAllowlist ? { senderAllowlist } : {}), + ...(topLevelReplies ? { topLevelReplies: true as const } : {}), + } + : undefined; +} + function shouldUseIsolatedQaSuiteScenarioWorkers(params: { scenarios: ReturnType["scenarios"]; concurrency: number; @@ -270,7 +303,11 @@ function shouldUseIsolatedQaSuiteScenarioWorkers(params: { return ( params.scenarios.length > 1 && (params.concurrency > 1 || - params.scenarios.some((scenario) => isQaMergePatchObject(scenario.gatewayConfigPatch))) + params.scenarios.some( + (scenario) => + isQaMergePatchObject(scenario.gatewayConfigPatch) || + (scenario.execution.kind === "flow" && scenario.execution.transportPolicy !== undefined), + )) ); } @@ -280,6 +317,8 @@ function scenarioRequiresIsolatedQaSuiteWorker(scenario: QaSeedScenario) { } return ( scenario.execution.suiteIsolation === "isolated" || + // Transport policy is fixed when the gateway starts; sharing it would leak routing rules. + scenario.execution.transportPolicy !== undefined || isQaMergePatchObject(scenario.gatewayConfigPatch) || scenario.gatewayRuntime !== undefined || (Array.isArray(scenario.plugins) && scenario.plugins.length > 0) || @@ -417,6 +456,7 @@ export { applyQaMergePatch, collectQaSuiteGatewayConfigPatch, collectQaSuiteGatewayRuntimeOptions, + collectQaSuiteTransportPolicy, collectQaSuitePluginIds, mapQaSuiteWithConcurrency, normalizeQaSuiteConcurrency, diff --git a/extensions/qa-lab/src/suite-test-helpers.ts b/extensions/qa-lab/src/suite-test-helpers.ts index ba4fbe2374c..7f98a95a206 100644 --- a/extensions/qa-lab/src/suite-test-helpers.ts +++ b/extensions/qa-lab/src/suite-test-helpers.ts @@ -1,4 +1,5 @@ // Qa Lab helper module supports suite test helpers behavior. +import type { QaTransportPolicy } from "./qa-transport.js"; import { readQaBootstrapScenarioCatalog } from "./scenario-catalog.js"; type QaSuiteTestScenario = ReturnType["scenarios"][number]; @@ -14,6 +15,7 @@ export function makeQaSuiteTestScenario( runtimeParityTier?: QaSuiteTestScenario["runtimeParityTier"]; suiteIsolation?: "isolated"; surface?: string; + transportPolicy?: QaTransportPolicy; } = {}, ): QaSuiteTestScenario { return { @@ -31,6 +33,7 @@ export function makeQaSuiteTestScenario( kind: "flow", ...(params.channel ? { channel: params.channel } : {}), ...(params.suiteIsolation ? { suiteIsolation: params.suiteIsolation } : {}), + ...(params.transportPolicy ? { transportPolicy: params.transportPolicy } : {}), ...(params.config ? { config: params.config } : {}), flow: { steps: [{ name: "noop", actions: [{ assert: "true" }] }] }, }, diff --git a/extensions/qa-lab/src/suite.test.ts b/extensions/qa-lab/src/suite.test.ts index 5b5433c5d13..3cea6ed9109 100644 --- a/extensions/qa-lab/src/suite.test.ts +++ b/extensions/qa-lab/src/suite.test.ts @@ -76,12 +76,41 @@ describe("qa suite", () => { channelDriver: "live", channelId: "telegram", outputDir: "/tmp/qa-output", + transportPolicy: { requireGroupMention: true }, state: {} as QaLabServerHandle["state"], transportId: "qa-channel", }), ).resolves.toMatchObject({ adapter }); expect(create).toHaveBeenCalledTimes(1); + expect(create).toHaveBeenCalledWith( + expect.objectContaining({ + adapterOptions: expect.objectContaining({ + transportPolicy: { requireGroupMention: true }, + }), + }), + ); + }); + + it("preserves caller-supplied transport policy without scenario metadata", async () => { + const adapter = { id: "telegram" } as QaTransportAdapter; + const create = vi.fn(async () => adapter); + + await qaSuiteProgressTesting.createQaSuiteTransportAdapter({ + adapterFactories: [{ id: "telegram", matches: () => true, create }], + adapterOptions: { transportPolicy: { topLevelReplies: true } }, + channelDriver: "live", + channelId: "telegram", + outputDir: "/tmp/qa-output", + state: {} as QaLabServerHandle["state"], + transportId: "qa-channel", + }); + + expect(create).toHaveBeenCalledWith( + expect.objectContaining({ + adapterOptions: { transportPolicy: { topLevelReplies: true } }, + }), + ); }); it("parses progress env booleans", () => { diff --git a/extensions/qa-lab/src/suite.ts b/extensions/qa-lab/src/suite.ts index 1111db7dd54..ddf99bb301d 100644 --- a/extensions/qa-lab/src/suite.ts +++ b/extensions/qa-lab/src/suite.ts @@ -71,6 +71,7 @@ import { applyQaMergePatch, collectQaSuiteGatewayConfigPatch, collectQaSuiteGatewayRuntimeOptions, + collectQaSuiteTransportPolicy, collectQaSuitePluginIds, mapQaSuiteWithConcurrency, normalizeQaSuiteConcurrency, @@ -124,6 +125,7 @@ async function createQaSuiteTransportAdapter(params: { channelDriverSelection?: OpenClawCrablineChannelDriverSelection | null; cleanupOnFailure?: () => Promise; outputDir: string; + transportPolicy?: NonNullable["transportPolicy"]; state: QaLabServerHandle["state"]; transportId: QaTransportId; }) { @@ -141,7 +143,17 @@ async function createQaSuiteTransportAdapter(params: { ? "crabline" : params.transportId, outputDir: params.outputDir, - adapterOptions: params.adapterOptions, + adapterOptions: { + ...params.adapterOptions, + ...(params.transportPolicy + ? { + transportPolicy: { + ...params.adapterOptions?.transportPolicy, + ...params.transportPolicy, + }, + } + : {}), + }, state: params.state, }, usesLiveAdapter ? params.adapterFactories : undefined, @@ -762,6 +774,7 @@ async function runQaRuntimeParitySuite(params: { adapterOptions: params.adapterOptions, cleanupOnFailure: ownsLab ? () => lab.stop() : undefined, outputDir: params.outputDir, + transportPolicy: collectQaSuiteTransportPolicy(params.selectedScenarios), state: lab.state, transportId: params.transportId, }); @@ -1604,6 +1617,7 @@ export async function runQaFlowSuite(params?: QaSuiteRunParams): Promise lab.stop() : undefined, outputDir, + transportPolicy: collectQaSuiteTransportPolicy(selectedScenarios), state: lab.state, transportId, }); diff --git a/extensions/qa-matrix/src/adapter.runtime.ts b/extensions/qa-matrix/src/adapter.runtime.ts index b19afb69ef8..7f154538baa 100644 --- a/extensions/qa-matrix/src/adapter.runtime.ts +++ b/extensions/qa-matrix/src/adapter.runtime.ts @@ -2,6 +2,7 @@ import { randomUUID } from "node:crypto"; import path from "node:path"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; +import { buildQaTarget } from "openclaw/plugin-sdk/qa-channel"; import type { QaRunnerCliRegistration } from "openclaw/plugin-sdk/qa-runner-runtime"; import { createMatrixQaClient, provisionMatrixQaRoom } from "./substrate/client.js"; import { buildMatrixQaConfig } from "./substrate/config.js"; @@ -24,6 +25,13 @@ const MATRIX_SHARED_FLOW_TOPOLOGY = { name: "OpenClaw Matrix QA", requireMention: true, }, + { + key: "secondary", + kind: "group", + members: ["driver", "observer", "sut"], + name: "OpenClaw Matrix QA Secondary Room", + requireMention: true, + }, { key: "driver-dm", kind: "dm", @@ -39,21 +47,21 @@ const MATRIX_SHARED_FLOW_TOPOLOGY = { ], } satisfies MatrixQaTopologySpec; -function resolveMatrixQaAdapterRoom(topology: MatrixQaProvisionedTopology, conversationId: string) { +function resolveMatrixQaAdapterRoom( + topology: MatrixQaProvisionedTopology, + conversation: { id: string; kind: "channel" | "direct" | "group" }, +) { return ( - topology.rooms.find((room) => room.key === conversationId || room.roomId === conversationId) ?? + topology.rooms.find( + (room) => room.key === conversation.id || room.roomId === conversation.id, + ) ?? + (conversation.kind === "direct" + ? topology.rooms.find((room) => room.kind === "dm") + : undefined) ?? topology.rooms.find((room) => room.roomId === topology.defaultRoomId)! ); } -function buildMatrixQaBusTarget(conversation: { - id: string; - kind: "channel" | "direct" | "group"; -}) { - const prefix = conversation.kind === "direct" ? "dm" : conversation.kind; - return `${prefix}:${conversation.id}`; -} - async function waitForMatrixChannelReady( gateway: Parameters[0]["gateway"], accountId: string, @@ -193,7 +201,10 @@ export async function createMatrixQaTransportAdapter( } const outbound = await context.messages.addOutboundMessage({ accountId, - to: buildMatrixQaBusTarget(logicalConversation), + to: buildQaTarget({ + chatType: logicalConversation.kind, + conversationId: logicalConversation.id, + }), senderId: event.sender, text, timestamp: event.originServerTs, @@ -226,7 +237,7 @@ export async function createMatrixQaTransportAdapter( } }, async sendInbound(input) { - const room = resolveMatrixQaAdapterRoom(provisioning.topology, input.conversation.id); + const room = resolveMatrixQaAdapterRoom(provisioning.topology, input.conversation); logicalConversationByRoomId.set(room.roomId, { id: input.conversation.id, kind: input.conversation.kind, diff --git a/extensions/qa-matrix/src/cli.ts b/extensions/qa-matrix/src/cli.ts index 7677dea9529..e554e8f156c 100644 --- a/extensions/qa-matrix/src/cli.ts +++ b/extensions/qa-matrix/src/cli.ts @@ -61,6 +61,13 @@ export const matrixQaAdapterFactory: NonNullable { const blockStreamingScenario = liveTesting.MATRIX_QA_SCENARIOS.find( (scenario) => scenario.id === "matrix-room-block-streaming", ); - const topLevelScenario = liveTesting.MATRIX_QA_SCENARIOS.find( - (scenario) => scenario.id === "matrix-top-level-reply-shape", + const threadScenario = liveTesting.MATRIX_QA_SCENARIOS.find( + (scenario) => scenario.id === "matrix-thread-root-preservation", ); expect(blockStreamingScenario).toBeDefined(); - expect(topLevelScenario).toBeDefined(); + expect(threadScenario).toBeDefined(); const pinnedSchedule = liveTesting.scheduleMatrixQaScenariosInCatalogOrder([ blockStreamingScenario!, @@ -247,7 +247,7 @@ describe("matrix live qa runtime", () => { expect(liveTesting.selectMatrixQaCanaryProviderMode(pinnedSchedule)).toBe("mock-openai"); const mixedSchedule = liveTesting.scheduleMatrixQaScenariosInCatalogOrder([ - topLevelScenario!, + threadScenario!, blockStreamingScenario!, ]); expect(liveTesting.selectMatrixQaCanaryProviderMode(mixedSchedule)).toBeUndefined(); @@ -521,51 +521,6 @@ describe("matrix live qa runtime", () => { expect(config.scenarios[0]?.config.blockStreaming).toBe(true); }); - it("preserves negative-scenario artifacts in the Matrix summary", () => { - const summary = liveTesting.buildMatrixQaSummary( - buildMatrixQaSummaryInput({ - scenarios: [ - { - id: "matrix-mention-gating", - title: "Matrix room message without mention does not trigger", - status: "pass", - details: "no reply", - artifacts: { - actorUserId: "@driver:matrix-qa.test", - driverEventId: "$driver", - expectedNoReplyWindowMs: 8_000, - token: "MATRIX_QA_NOMENTION_TOKEN", - triggerBody: "reply with only this exact marker: MATRIX_QA_NOMENTION_TOKEN", - }, - }, - ], - timings: { - scenarios: [ - { - durationMs: 80, - gatewayBootMs: 0, - gatewayRestartMs: 0, - id: "matrix-mention-gating", - title: "Matrix room message without mention does not trigger", - transportInterruptMs: 0, - }, - ], - totalMs: 905, - }, - }), - ); - expect(summary.counts.total).toBe(2); - expect(summary.counts.passed).toBe(2); - expect(summary.counts.failed).toBe(0); - expect(summary.scenarios[0]?.id).toBe("matrix-mention-gating"); - expect(summary.scenarios[0]?.artifacts?.actorUserId).toBe("@driver:matrix-qa.test"); - expect(summary.scenarios[0]?.artifacts?.expectedNoReplyWindowMs).toBe(8_000); - expect(summary.scenarios[0]?.artifacts?.triggerBody).toBe( - "reply with only this exact marker: MATRIX_QA_NOMENTION_TOKEN", - ); - expect(summary.timings.totalMs).toBe(905); - }); - it("keeps failing Matrix scenario details and timings complete in summary + report output", () => { const summary = liveTesting.buildMatrixQaSummary( buildMatrixQaSummaryInput({ @@ -628,7 +583,7 @@ describe("matrix live qa runtime", () => { it("groups Matrix scenario execution by gateway config while preserving tail scenarios", () => { const scenarios = liveTesting.findMatrixQaScenarios([ - "matrix-top-level-reply-shape", + "matrix-thread-root-preservation", "matrix-e2ee-cli-encryption-setup-multi-account", "matrix-reaction-notification", "matrix-e2ee-cli-setup-then-gateway-reply", @@ -641,7 +596,7 @@ describe("matrix live qa runtime", () => { .scheduleMatrixQaScenariosInCatalogOrder(scenarios) .map(({ scenario }) => scenario.id), ).toEqual([ - "matrix-top-level-reply-shape", + "matrix-thread-root-preservation", "matrix-reaction-notification", "matrix-e2ee-cli-self-verification", "matrix-e2ee-cli-encryption-setup-multi-account", diff --git a/extensions/qa-matrix/src/runners/contract/scenario-catalog.ts b/extensions/qa-matrix/src/runners/contract/scenario-catalog.ts index d271871778d..8a2e60289f6 100644 --- a/extensions/qa-matrix/src/runners/contract/scenario-catalog.ts +++ b/extensions/qa-matrix/src/runners/contract/scenario-catalog.ts @@ -17,7 +17,6 @@ import { type MatrixQaScenarioId = | "matrix-thread-root-preservation" | "matrix-thread-nested-reply-shape" - | "matrix-top-level-reply-shape" | "matrix-room-partial-streaming-preview" | "matrix-room-quiet-streaming-preview" | "matrix-room-tool-progress-preview" @@ -32,9 +31,7 @@ type MatrixQaScenarioId = | "matrix-voice-preflight-mention" | "matrix-attachment-only-ignored" | "matrix-unsupported-media-safe" - | "matrix-dm-reply-shape" | "matrix-room-autojoin-invite" - | "matrix-secondary-room-reply" | "matrix-secondary-room-open-trigger" | "matrix-reaction-notification" | "matrix-reaction-threaded" @@ -50,7 +47,6 @@ type MatrixQaScenarioId = | "matrix-stale-sync-replay-dedupe" | "matrix-room-membership-loss" | "matrix-homeserver-restart-resume" - | "matrix-mention-gating" | "matrix-allowbots-default-block" | "matrix-allowbots-true-unmentioned-open-room" | "matrix-allowbots-mentions-mentioned-room" @@ -61,9 +57,6 @@ type MatrixQaScenarioId = | "matrix-allowbots-self-sender-ignored" | "matrix-mxid-prefixed-command-block" | "matrix-mention-metadata-spoof-block" - | "matrix-observer-allowlist-override" - | "matrix-allowlist-block" - | "matrix-multi-actor-ordering" | "matrix-inbound-edit-ignored" | "matrix-inbound-edit-no-duplicate-trigger" | "matrix-e2ee-basic-reply" @@ -329,12 +322,6 @@ export const MATRIX_QA_SCENARIOS: MatrixQaScenarioDefinition[] = [ timeoutMs: 60_000, title: "Matrix nested threaded replies keep fallback replies on the root event", }, - { - id: "matrix-top-level-reply-shape", - standardId: "top-level-reply-shape", - timeoutMs: 45_000, - title: "Matrix top-level reply keeps replyToMode off", - }, { id: "matrix-room-partial-streaming-preview", timeoutMs: 45_000, @@ -471,12 +458,6 @@ export const MATRIX_QA_SCENARIOS: MatrixQaScenarioDefinition[] = [ title: "Matrix unsupported media attachments do not block caption replies", topology: MATRIX_QA_MEDIA_ROOM_TOPOLOGY, }, - { - id: "matrix-dm-reply-shape", - timeoutMs: 45_000, - title: "Matrix DM reply stays top-level without a mention", - topology: MATRIX_QA_DRIVER_DM_TOPOLOGY, - }, { id: "matrix-room-autojoin-invite", timeoutMs: 60_000, @@ -486,12 +467,6 @@ export const MATRIX_QA_SCENARIOS: MatrixQaScenarioDefinition[] = [ groupPolicy: "open", }, }, - { - id: "matrix-secondary-room-reply", - timeoutMs: 45_000, - title: "Matrix secondary room reply stays scoped to that room", - topology: MATRIX_QA_SECONDARY_ROOM_TOPOLOGY, - }, { id: "matrix-secondary-room-open-trigger", timeoutMs: 45_000, @@ -587,12 +562,6 @@ export const MATRIX_QA_SCENARIOS: MatrixQaScenarioDefinition[] = [ title: "Matrix lane resumes after homeserver restart", topology: MATRIX_QA_HOMESERVER_ROOM_TOPOLOGY, }, - { - id: "matrix-mention-gating", - standardId: "mention-gating", - timeoutMs: 8_000, - title: "Matrix room message without mention does not trigger", - }, { id: "matrix-allowbots-default-block", timeoutMs: 8_000, @@ -710,25 +679,6 @@ export const MATRIX_QA_SCENARIOS: MatrixQaScenarioDefinition[] = [ timeoutMs: 8_000, title: "Matrix metadata-only mention spoof does not trigger", }, - { - id: "matrix-observer-allowlist-override", - timeoutMs: 45_000, - title: "Matrix sender allowlist override lets observer messages trigger replies", - configOverrides: { - groupAllowRoles: ["driver", "observer"], - }, - }, - { - id: "matrix-allowlist-block", - standardId: "allowlist-block", - timeoutMs: 8_000, - title: "Matrix sender allowlist blocks observer replies", - }, - { - id: "matrix-multi-actor-ordering", - timeoutMs: 60_000, - title: "Matrix blocked observer traffic does not poison later driver replies", - }, { id: "matrix-inbound-edit-ignored", timeoutMs: 8_000, @@ -1113,14 +1063,11 @@ export const MATRIX_QA_PROFILE_NAMES: readonly MatrixQaProfile[] = [ ] as const; const MATRIX_QA_FAST_PROFILE_SCENARIO_IDS = [ - "matrix-top-level-reply-shape", "matrix-reaction-notification", "matrix-approval-exec-metadata-single-event", "matrix-approval-exec-metadata-chunked", - "matrix-mention-gating", "matrix-allowbots-default-block", "matrix-allowbots-mentions-mentioned-room", - "matrix-allowlist-block", "matrix-e2ee-basic-reply", ] satisfies MatrixQaScenarioId[]; diff --git a/extensions/qa-matrix/src/runners/contract/scenario-runtime-room.ts b/extensions/qa-matrix/src/runners/contract/scenario-runtime-room.ts index 35057e9588a..9462f557d2e 100644 --- a/extensions/qa-matrix/src/runners/contract/scenario-runtime-room.ts +++ b/extensions/qa-matrix/src/runners/contract/scenario-runtime-room.ts @@ -35,7 +35,6 @@ import { isMatrixQaExactMarkerReply, isMatrixQaMessageLikeKind, MATRIX_QA_TOOL_PROGRESS_TASK_FILENAME, - primeMatrixQaActorCursor, primeMatrixQaDriverScenarioClient, resolveMatrixQaNoReplyWindowMs, runAssertedDriverTopLevelScenario, @@ -244,77 +243,6 @@ export async function runThreadNestedReplyShapeScenario(context: MatrixQaScenari } satisfies MatrixQaScenarioExecution; } -export async function runTopLevelReplyShapeScenario(context: MatrixQaScenarioContext) { - const result = await runAssertedDriverTopLevelScenario({ - context, - label: "top-level reply", - tokenPrefix: "MATRIX_QA_TOPLEVEL", - }); - return { - artifacts: { - driverEventId: result.driverEventId, - reply: result.reply, - token: result.token, - }, - details: [ - `driver event: ${result.driverEventId}`, - ...buildMatrixReplyDetails("reply", result.reply), - ].join("\n"), - } satisfies MatrixQaScenarioExecution; -} - -export async function runObserverAllowlistOverrideScenario(context: MatrixQaScenarioContext) { - const { client, startSince } = await primeMatrixQaActorCursor({ - accessToken: context.observerAccessToken, - actorId: "observer", - baseUrl: context.baseUrl, - observedEvents: context.observedEvents, - syncState: context.syncState, - syncStreams: context.syncStreams, - }); - const token = buildMatrixQaToken("MATRIX_QA_OBSERVER_ALLOWLIST"); - const body = buildMentionPrompt(context.sutUserId, token); - const driverEventId = await client.sendTextMessage({ - body, - mentionUserIds: [context.sutUserId], - roomId: context.roomId, - }); - const matched = await client.waitForRoomEvent({ - observedEvents: context.observedEvents, - predicate: (event) => - isMatrixQaExactMarkerReply(event, { - roomId: context.roomId, - sutUserId: context.sutUserId, - token, - }) && event.relatesTo === undefined, - roomId: context.roomId, - since: startSince, - timeoutMs: context.timeoutMs, - }); - advanceMatrixQaActorCursor({ - actorId: "observer", - syncState: context.syncState, - nextSince: matched.since, - startSince, - }); - const reply = buildMatrixReplyArtifact(matched.event, token); - assertTopLevelReplyArtifact("observer allowlist reply", reply); - return { - artifacts: { - actorUserId: context.observerUserId, - driverEventId, - reply, - token, - triggerBody: body, - }, - details: [ - `trigger sender: ${context.observerUserId}`, - `driver event: ${driverEventId}`, - ...buildMatrixReplyDetails("reply", reply), - ].join("\n"), - } satisfies MatrixQaScenarioExecution; -} - export async function runQuietStreamingPreviewScenario(context: MatrixQaScenarioContext) { return runMatrixStreamingPreviewScenario(context, { expectedPreviewKind: "notice", diff --git a/extensions/qa-matrix/src/runners/contract/scenario-runtime.ts b/extensions/qa-matrix/src/runners/contract/scenario-runtime.ts index 8ba7a3e83b1..e544edd8035 100644 --- a/extensions/qa-matrix/src/runners/contract/scenario-runtime.ts +++ b/extensions/qa-matrix/src/runners/contract/scenario-runtime.ts @@ -1,9 +1,5 @@ // Qa Matrix plugin module implements scenario runtime behavior. -import { - MATRIX_QA_DRIVER_DM_ROOM_KEY, - MATRIX_QA_SECONDARY_ROOM_KEY, - type MatrixQaScenarioDefinition, -} from "./scenario-catalog.js"; +import { MATRIX_QA_SECONDARY_ROOM_KEY, type MatrixQaScenarioDefinition } from "./scenario-catalog.js"; import { runAllowBotsDefaultBlockScenario, runAllowBotsMentionsDmUnmentionedScenario, @@ -88,7 +84,6 @@ import { runBlockStreamingScenario, runMatrixQaCanary, runMembershipLossScenario, - runObserverAllowlistOverrideScenario, runPartialStreamingPreviewScenario, runQuietStreamingPreviewScenario, runReactionThreadedScenario, @@ -100,7 +95,6 @@ import { runToolProgressMentionSafetyScenario, runToolProgressPreviewOptOutScenario, runToolProgressPreviewScenario, - runTopLevelReplyShapeScenario, } from "./scenario-runtime-room.js"; import { buildExactMarkerPrompt, @@ -109,7 +103,6 @@ import { buildMatrixReplyDetails, buildMentionPrompt, readMatrixQaSyncCursor, - resolveMatrixQaNoReplyWindowMs, runNoReplyExpectedScenario, runTopologyScopedTopLevelScenario, writeMatrixQaSyncCursor, @@ -172,32 +165,6 @@ async function runNoReplyScenario(params: { }); } -async function runMultiActorOrderingScenario(context: MatrixQaScenarioContext) { - const blockedToken = buildMatrixQaToken("MATRIX_QA_MULTI_BLOCKED"); - const blocked = await runNoReplyScenario({ - accessToken: context.observerAccessToken, - actorId: "observer", - actorUserId: context.observerUserId, - body: buildMentionPrompt(context.sutUserId, blockedToken), - mentionUserIds: [context.sutUserId], - context, - timeoutMs: resolveMatrixQaNoReplyWindowMs(context.timeoutMs), - token: blockedToken, - }); - const accepted = await runDriverTopologyScopedScenario({ - context, - roomKey: context.topology.defaultRoomKey, - tokenPrefix: "MATRIX_QA_MULTI_DRIVER", - }); - return { - artifacts: { - accepted: accepted.artifacts ?? {}, - blocked: blocked.artifacts ?? {}, - }, - details: [blocked.details, accepted.details].join("\n"), - } satisfies MatrixQaScenarioExecution; -} - export async function runMatrixQaScenario( scenario: MatrixQaScenarioDefinition, context: MatrixQaScenarioContext, @@ -207,8 +174,6 @@ export async function runMatrixQaScenario( return await runThreadRootPreservationScenario(context); case "matrix-thread-nested-reply-shape": return await runThreadNestedReplyShapeScenario(context); - case "matrix-top-level-reply-shape": - return await runTopLevelReplyShapeScenario(context); case "matrix-room-partial-streaming-preview": return await runPartialStreamingPreviewScenario(context); case "matrix-room-quiet-streaming-preview": @@ -237,21 +202,8 @@ export async function runMatrixQaScenario( return await runAttachmentOnlyIgnoredScenario(context); case "matrix-unsupported-media-safe": return await runUnsupportedMediaSafeScenario(context); - case "matrix-dm-reply-shape": - return await runDriverTopologyScopedScenario({ - context, - roomKey: MATRIX_QA_DRIVER_DM_ROOM_KEY, - tokenPrefix: "MATRIX_QA_DM", - withMention: false, - }); case "matrix-room-autojoin-invite": return await runRoomAutoJoinInviteScenario(context); - case "matrix-secondary-room-reply": - return await runDriverTopologyScopedScenario({ - context, - roomKey: MATRIX_QA_SECONDARY_ROOM_KEY, - tokenPrefix: "MATRIX_QA_SECONDARY", - }); case "matrix-secondary-room-open-trigger": return await runDriverTopologyScopedScenario({ context, @@ -287,17 +239,6 @@ export async function runMatrixQaScenario( return await runMembershipLossScenario(context); case "matrix-homeserver-restart-resume": return await runHomeserverRestartResumeScenario(context); - case "matrix-mention-gating": { - const token = buildMatrixQaToken("MATRIX_QA_NOMENTION"); - return await runNoReplyScenario({ - accessToken: context.driverAccessToken, - actorId: "driver", - actorUserId: context.driverUserId, - body: buildExactMarkerPrompt(token), - context, - token, - }); - } case "matrix-allowbots-default-block": return await runAllowBotsDefaultBlockScenario(context); case "matrix-allowbots-true-unmentioned-open-room": @@ -338,22 +279,6 @@ export async function runMatrixQaScenario( token, }); } - case "matrix-observer-allowlist-override": - return await runObserverAllowlistOverrideScenario(context); - case "matrix-allowlist-block": { - const token = buildMatrixQaToken("MATRIX_QA_ALLOWLIST"); - return await runNoReplyScenario({ - accessToken: context.observerAccessToken, - actorId: "observer", - actorUserId: context.observerUserId, - body: buildMentionPrompt(context.sutUserId, token), - mentionUserIds: [context.sutUserId], - context, - token, - }); - } - case "matrix-multi-actor-ordering": - return await runMultiActorOrderingScenario(context); case "matrix-inbound-edit-ignored": return await runInboundEditIgnoredScenario(context); case "matrix-inbound-edit-no-duplicate-trigger": diff --git a/extensions/qa-matrix/src/runners/contract/scenarios.test.ts b/extensions/qa-matrix/src/runners/contract/scenarios.test.ts index af157d9a7e0..913c382165e 100644 --- a/extensions/qa-matrix/src/runners/contract/scenarios.test.ts +++ b/extensions/qa-matrix/src/runners/contract/scenarios.test.ts @@ -383,7 +383,6 @@ describe("matrix live qa scenarios", () => { expect(scenarioTesting.findMatrixQaScenarios().map((scenario) => scenario.id)).toEqual([ "matrix-thread-root-preservation", "matrix-thread-nested-reply-shape", - "matrix-top-level-reply-shape", "matrix-room-partial-streaming-preview", "matrix-room-quiet-streaming-preview", "matrix-room-tool-progress-preview", @@ -397,9 +396,7 @@ describe("matrix live qa scenarios", () => { "matrix-voice-preflight-mention", "matrix-attachment-only-ignored", "matrix-unsupported-media-safe", - "matrix-dm-reply-shape", "matrix-room-autojoin-invite", - "matrix-secondary-room-reply", "matrix-secondary-room-open-trigger", "matrix-reaction-notification", "matrix-reaction-threaded", @@ -415,7 +412,6 @@ describe("matrix live qa scenarios", () => { "matrix-stale-sync-replay-dedupe", "matrix-room-membership-loss", "matrix-homeserver-restart-resume", - "matrix-mention-gating", "matrix-allowbots-default-block", "matrix-allowbots-true-unmentioned-open-room", "matrix-allowbots-mentions-mentioned-room", @@ -426,9 +422,6 @@ describe("matrix live qa scenarios", () => { "matrix-allowbots-self-sender-ignored", "matrix-mxid-prefixed-command-block", "matrix-mention-metadata-spoof-block", - "matrix-observer-allowlist-override", - "matrix-allowlist-block", - "matrix-multi-actor-ordering", "matrix-inbound-edit-ignored", "matrix-inbound-edit-no-duplicate-trigger", "matrix-e2ee-basic-reply", @@ -622,14 +615,11 @@ describe("matrix live qa scenarios", () => { expect( scenarioTesting.findMatrixQaScenarios(undefined, "fast").map((scenario) => scenario.id), ).toEqual([ - "matrix-top-level-reply-shape", "matrix-reaction-notification", "matrix-approval-exec-metadata-single-event", "matrix-approval-exec-metadata-chunked", - "matrix-mention-gating", "matrix-allowbots-default-block", "matrix-allowbots-mentions-mentioned-room", - "matrix-allowlist-block", "matrix-e2ee-basic-reply", ]); }); @@ -933,24 +923,26 @@ describe("matrix live qa scenarios", () => { it("fails when any requested Matrix scenario id is unknown", () => { expect(() => - scenarioTesting.findMatrixQaScenarios(["matrix-top-level-reply-shape", "typo-scenario"]), + scenarioTesting.findMatrixQaScenarios(["matrix-thread-root-preservation", "typo-scenario"]), ).toThrow("unknown Matrix QA scenario id(s): typo-scenario"); }); - it("covers the baseline live transport contract plus Matrix-specific extras", () => { + it("keeps only Matrix-native standards in the imperative catalog", () => { expect(scenarioTesting.MATRIX_QA_STANDARD_SCENARIO_IDS).toEqual([ "canary", - "top-level-reply-shape", "reaction-observation", - "mention-gating", - "allowlist-block", ]); expect( findMissingLiveTransportStandardScenarios({ coveredStandardScenarioIds: scenarioTesting.MATRIX_QA_STANDARD_SCENARIO_IDS, expectedStandardScenarioIds: LIVE_TRANSPORT_BASELINE_STANDARD_SCENARIO_IDS, }), - ).toStrictEqual(["restart-resume"]); + ).toStrictEqual([ + "mention-gating", + "allowlist-block", + "top-level-reply-shape", + "restart-resume", + ]); }); it("merges default and scenario-requested Matrix topology once per run", () => { @@ -960,10 +952,10 @@ describe("matrix live qa scenarios", () => { scenarios: [ MATRIX_QA_SCENARIOS[0], { - id: "matrix-secondary-room-reply", + id: "matrix-secondary-room-open-trigger", standardId: "canary", timeoutMs: 60_000, - title: "Matrix restart resume", + title: "Matrix secondary topology", topology: { defaultRoomKey: "main", rooms: [ @@ -1019,7 +1011,7 @@ describe("matrix live qa scenarios", () => { defaultRoomName: "OpenClaw Matrix QA run", scenarios: [ { - id: "matrix-top-level-reply-shape", + id: "matrix-thread-root-preservation", timeoutMs: 60_000, title: "A", topology: { @@ -1363,132 +1355,6 @@ describe("matrix live qa scenarios", () => { ).toBe("!main:matrix-qa.test"); }); - it("primes the observer sync cursor instead of reusing the driver's cursor", async () => { - const primeRoom = vi.fn().mockResolvedValue("observer-sync-start"); - const sendTextMessage = vi.fn().mockResolvedValue("$observer-trigger"); - const waitForOptionalRoomEvent = vi.fn().mockImplementation(async (params) => { - expect(params.since).toBe("observer-sync-start"); - return { - matched: false, - since: "observer-sync-next", - }; - }); - - createMatrixQaClient.mockReturnValue({ - primeRoom, - sendTextMessage, - waitForOptionalRoomEvent, - }); - - const scenario = requireMatrixQaScenario("matrix-allowlist-block"); - - const syncState = { - driver: "driver-sync-next", - }; - - const result = await runMatrixQaScenario(scenario, { - baseUrl: "http://127.0.0.1:28008/", - canary: undefined, - driverAccessToken: "driver-token", - driverUserId: "@driver:matrix-qa.test", - observedEvents: [], - observerAccessToken: "observer-token", - observerUserId: "@observer:matrix-qa.test", - roomId: "!room:matrix-qa.test", - restartGateway: undefined, - syncState, - sutAccessToken: "sut-token", - sutUserId: "@sut:matrix-qa.test", - timeoutMs: 8_000, - topology: { - defaultRoomId: "!room:matrix-qa.test", - defaultRoomKey: "main", - rooms: [], - }, - }); - const artifacts = result.artifacts as Record; - expect(artifacts.actorUserId).toBe("@observer:matrix-qa.test"); - expect(artifacts.expectedNoReplyWindowMs).toBe(8_000); - - expect(createMatrixQaClient).toHaveBeenCalledWith({ - accessToken: "observer-token", - baseUrl: "http://127.0.0.1:28008/", - }); - expect(primeRoom).toHaveBeenCalledTimes(1); - expect(sendTextMessage).toHaveBeenCalledTimes(1); - expect(waitForOptionalRoomEvent).toHaveBeenCalledTimes(1); - expect(syncState).toEqual({ - driver: "driver-sync-next", - observer: "observer-sync-next", - }); - }); - - it("allows observer messages when the sender allowlist override includes them", async () => { - const primeRoom = vi.fn().mockResolvedValue("observer-sync-start"); - const sendTextMessage = vi.fn().mockResolvedValue("$observer-allow-trigger"); - const waitForRoomEvent = vi.fn().mockImplementation(async () => ({ - event: { - kind: "message", - roomId: "!room:matrix-qa.test", - eventId: "$sut-reply", - sender: "@sut:matrix-qa.test", - type: "m.room.message", - body: mockMessageBody(sendTextMessage, "sendTextMessage").replace( - "@sut:matrix-qa.test reply with only this exact marker: ", - "", - ), - }, - since: "observer-sync-next", - })); - - createMatrixQaClient.mockReturnValue({ - primeRoom, - sendTextMessage, - waitForRoomEvent, - }); - - const scenario = requireMatrixQaScenario("matrix-observer-allowlist-override"); - - const result = await runMatrixQaScenario(scenario, { - baseUrl: "http://127.0.0.1:28008/", - canary: undefined, - driverAccessToken: "driver-token", - driverUserId: "@driver:matrix-qa.test", - observedEvents: [], - observerAccessToken: "observer-token", - observerUserId: "@observer:matrix-qa.test", - roomId: "!room:matrix-qa.test", - restartGateway: undefined, - syncState: {}, - sutAccessToken: "sut-token", - sutUserId: "@sut:matrix-qa.test", - timeoutMs: 8_000, - topology: { - defaultRoomId: "!room:matrix-qa.test", - defaultRoomKey: "main", - rooms: [], - }, - }); - const artifacts = result.artifacts as { - actorUserId?: unknown; - driverEventId?: unknown; - reply?: { tokenMatched?: unknown }; - }; - expect(artifacts.actorUserId).toBe("@observer:matrix-qa.test"); - expect(artifacts.driverEventId).toBe("$observer-allow-trigger"); - expect(artifacts.reply?.tokenMatched).toBe(true); - - expect(createMatrixQaClient).toHaveBeenCalledWith({ - accessToken: "observer-token", - baseUrl: "http://127.0.0.1:28008/", - }); - expectSentTextMessage(sendTextMessage, { - bodyIncludes: "@sut:matrix-qa.test reply with only this exact marker:", - mentionUserIds: ["@sut:matrix-qa.test"], - roomId: "!room:matrix-qa.test", - }); - }); - it("runs mentioned allowBots=mentions room traffic through the observer bot account", async () => { const primeRoom = vi.fn().mockResolvedValue("observer-sync-start"); const sendTextMessage = vi.fn().mockResolvedValue("$observer-bot-trigger"); @@ -2407,84 +2273,6 @@ describe("matrix live qa scenarios", () => { } }); - it("runs the DM scenario against the provisioned DM room without a mention", async () => { - const primeRoom = vi.fn().mockResolvedValue("driver-sync-start"); - const sendTextMessage = vi.fn().mockResolvedValue("$dm-trigger"); - const waitForRoomEvent = vi.fn().mockImplementation(async () => ({ - event: { - roomId: "!dm:matrix-qa.test", - eventId: "$sut-reply", - sender: "@sut:matrix-qa.test", - type: "m.room.message", - body: mockMessageBody(sendTextMessage, "sendTextMessage").replace( - "reply with only this exact marker: ", - "", - ), - }, - since: "driver-sync-next", - })); - - createMatrixQaClient.mockReturnValue({ - primeRoom, - sendTextMessage, - waitForRoomEvent, - }); - - const scenario = requireMatrixQaScenario("matrix-dm-reply-shape"); - - const result = await runMatrixQaScenario(scenario, { - baseUrl: "http://127.0.0.1:28008/", - canary: undefined, - driverAccessToken: "driver-token", - driverUserId: "@driver:matrix-qa.test", - observedEvents: [], - observerAccessToken: "observer-token", - observerUserId: "@observer:matrix-qa.test", - roomId: "!main:matrix-qa.test", - restartGateway: undefined, - syncState: {}, - sutAccessToken: "sut-token", - sutUserId: "@sut:matrix-qa.test", - timeoutMs: 8_000, - topology: { - defaultRoomId: "!main:matrix-qa.test", - defaultRoomKey: "main", - rooms: [ - { - key: "main", - kind: "group", - memberRoles: ["driver", "observer", "sut"], - memberUserIds: [ - "@driver:matrix-qa.test", - "@observer:matrix-qa.test", - "@sut:matrix-qa.test", - ], - name: "Main", - requireMention: true, - roomId: "!main:matrix-qa.test", - }, - { - key: scenarioTesting.MATRIX_QA_DRIVER_DM_ROOM_KEY, - kind: "dm", - memberRoles: ["driver", "sut"], - memberUserIds: ["@driver:matrix-qa.test", "@sut:matrix-qa.test"], - name: "DM", - requireMention: false, - roomId: "!dm:matrix-qa.test", - }, - ], - }, - }); - const artifacts = result.artifacts as { actorUserId?: unknown }; - expect(artifacts.actorUserId).toBe("@driver:matrix-qa.test"); - - expectSentTextMessage(sendTextMessage, { - bodyIncludes: "reply with only this exact marker:", - roomId: "!dm:matrix-qa.test", - }); - expect(mockObjectArg(waitForRoomEvent, "waitForRoomEvent").roomId).toBe("!dm:matrix-qa.test"); - }); - it("captures quiet preview notices before the finalized Matrix reply", async () => { const primeRoom = vi.fn().mockResolvedValue("driver-sync-start"); const sendTextMessage = vi.fn().mockResolvedValue("$quiet-stream-trigger"); @@ -4464,91 +4252,6 @@ describe("matrix live qa scenarios", () => { }); }); - it("runs the secondary-room scenario against the provisioned secondary room", async () => { - const primeRoom = vi.fn().mockResolvedValue("driver-sync-start"); - const sendTextMessage = vi.fn().mockResolvedValue("$secondary-trigger"); - const waitForRoomEvent = vi.fn().mockImplementation(async () => ({ - event: { - roomId: "!secondary:matrix-qa.test", - eventId: "$sut-reply", - sender: "@sut:matrix-qa.test", - type: "m.room.message", - body: mockMessageBody(sendTextMessage, "sendTextMessage").replace( - "@sut:matrix-qa.test reply with only this exact marker: ", - "", - ), - }, - since: "driver-sync-next", - })); - - createMatrixQaClient.mockReturnValue({ - primeRoom, - sendTextMessage, - waitForRoomEvent, - }); - - const scenario = requireMatrixQaScenario("matrix-secondary-room-reply"); - - const result = await runMatrixQaScenario(scenario, { - baseUrl: "http://127.0.0.1:28008/", - canary: undefined, - driverAccessToken: "driver-token", - driverUserId: "@driver:matrix-qa.test", - observedEvents: [], - observerAccessToken: "observer-token", - observerUserId: "@observer:matrix-qa.test", - roomId: "!main:matrix-qa.test", - restartGateway: undefined, - syncState: {}, - sutAccessToken: "sut-token", - sutUserId: "@sut:matrix-qa.test", - timeoutMs: 8_000, - topology: { - defaultRoomId: "!main:matrix-qa.test", - defaultRoomKey: "main", - rooms: [ - { - key: "main", - kind: "group", - memberRoles: ["driver", "observer", "sut"], - memberUserIds: [ - "@driver:matrix-qa.test", - "@observer:matrix-qa.test", - "@sut:matrix-qa.test", - ], - name: "Main", - requireMention: true, - roomId: "!main:matrix-qa.test", - }, - { - key: scenarioTesting.MATRIX_QA_SECONDARY_ROOM_KEY, - kind: "group", - memberRoles: ["driver", "observer", "sut"], - memberUserIds: [ - "@driver:matrix-qa.test", - "@observer:matrix-qa.test", - "@sut:matrix-qa.test", - ], - name: "Secondary", - requireMention: true, - roomId: "!secondary:matrix-qa.test", - }, - ], - }, - }); - const artifacts = result.artifacts as { actorUserId?: unknown }; - expect(artifacts.actorUserId).toBe("@driver:matrix-qa.test"); - - expectSentTextMessage(sendTextMessage, { - bodyIncludes: "@sut:matrix-qa.test", - mentionUserIds: ["@sut:matrix-qa.test"], - roomId: "!secondary:matrix-qa.test", - }); - expect(mockObjectArg(waitForRoomEvent, "waitForRoomEvent").roomId).toBe( - "!secondary:matrix-qa.test", - ); - }); - it("ignores stale E2EE replies when checking a verification notice", async () => { let noticeToken = ""; const sendNoticeMessage = vi.fn().mockImplementation(async ({ body }) => { diff --git a/extensions/qa-matrix/src/substrate/sync.test.ts b/extensions/qa-matrix/src/substrate/sync.test.ts index 893e4089dc2..6da293a66d0 100644 --- a/extensions/qa-matrix/src/substrate/sync.test.ts +++ b/extensions/qa-matrix/src/substrate/sync.test.ts @@ -246,6 +246,64 @@ describe("matrix sync helpers", () => { expect(calls).toBe(1); }); + it("keeps independent cursors for events observed while polling another room", async () => { + let calls = 0; + const fetchImpl: typeof fetch = async () => { + calls += 1; + return new Response( + JSON.stringify({ + next_batch: "next-batch-2", + rooms: { + join: { + "!main:matrix-qa.test": { + timeline: { + events: [ + { + event_id: "$main-reply", + sender: "@sut:matrix-qa.test", + type: "m.room.message", + content: { body: "main reply", msgtype: "m.text" }, + }, + ], + }, + }, + }, + }, + }), + { status: 200, headers: { "content-type": "application/json" } }, + ); + }; + const observer = createMatrixQaRoomObserver({ + accessToken: "token", + baseUrl: "http://127.0.0.1:28008/", + fetchImpl, + observedEvents: [], + since: "start-batch", + }); + + const nowSpy = vi.spyOn(Date, "now").mockReturnValueOnce(0).mockReturnValue(1); + try { + await expect( + observer.waitForOptionalRoomEvent({ + predicate: (event) => event.sender === "@sut:matrix-qa.test", + roomId: "!secondary:matrix-qa.test", + timeoutMs: 1, + }), + ).resolves.toEqual({ matched: false, since: "next-batch-2" }); + } finally { + nowSpy.mockRestore(); + } + + await expect( + observer.waitForRoomEvent({ + predicate: (event) => event.eventId === "$main-reply", + roomId: "!main:matrix-qa.test", + timeoutMs: 1_000, + }), + ).resolves.toMatchObject({ event: { eventId: "$main-reply" } }); + expect(calls).toBe(1); + }); + it("shares one in-flight /sync poll across concurrent waits", async () => { let calls = 0; let markFetchStarted: () => void = () => {}; diff --git a/extensions/qa-matrix/src/substrate/sync.ts b/extensions/qa-matrix/src/substrate/sync.ts index 72a7ccad6a6..a8969e43935 100644 --- a/extensions/qa-matrix/src/substrate/sync.ts +++ b/extensions/qa-matrix/src/substrate/sync.ts @@ -56,7 +56,7 @@ export type MatrixQaRoomObserver = { }; type MatrixQaRoomObserverState = { - cursorIndex: number; + cursorIndexes: Map; events: MatrixQaObservedEvent[]; pollPromise?: Promise; since?: string; @@ -128,7 +128,7 @@ export function createMatrixQaRoomObserver( }, ): MatrixQaRoomObserver { const roomObserver: MatrixQaRoomObserverState = { - cursorIndex: 0, + cursorIndexes: new Map(), events: [], since: params.since, }; @@ -144,7 +144,7 @@ export function createMatrixQaRoomObserver( async waitForOptionalRoomEvent(waitParams) { const startSince = await this.prime(); const startedAt = Date.now(); - let cursorIndex = roomObserver.cursorIndex; + let cursorIndex = roomObserver.cursorIndexes.get(waitParams.roomId) ?? 0; let didPoll = false; while (true) { const matched = findMatrixQaObservedEventMatch({ @@ -154,7 +154,13 @@ export function createMatrixQaRoomObserver( roomId: waitParams.roomId, }); if (matched) { - roomObserver.cursorIndex = Math.max(roomObserver.cursorIndex, matched.nextCursorIndex); + roomObserver.cursorIndexes.set( + waitParams.roomId, + Math.max( + roomObserver.cursorIndexes.get(waitParams.roomId) ?? 0, + matched.nextCursorIndex, + ), + ); return { event: matched.event, matched: true, @@ -164,7 +170,10 @@ export function createMatrixQaRoomObserver( const elapsedMs = Date.now() - startedAt; if (elapsedMs >= waitParams.timeoutMs && (didPoll || waitParams.timeoutMs <= 0)) { - roomObserver.cursorIndex = Math.max(roomObserver.cursorIndex, cursorIndex); + roomObserver.cursorIndexes.set( + waitParams.roomId, + Math.max(roomObserver.cursorIndexes.get(waitParams.roomId) ?? 0, cursorIndex), + ); return { matched: false, since: roomObserver.since ?? startSince, diff --git a/qa/scenarios/channels/channel-canary.yaml b/qa/scenarios/channels/channel-canary.yaml new file mode 100644 index 00000000000..f382ef3860c --- /dev/null +++ b/qa/scenarios/channels/channel-canary.yaml @@ -0,0 +1,63 @@ +title: Channel transport canary + +scenario: + id: channel-canary + surface: channels + coverage: + primary: + - channels.group-messages + secondary: + - channels.qa-channel + - runtime.delivery + objective: Verify the selected channel transport can deliver one mention-triggered group reply through the canonical QA host. + successCriteria: + - One mentioned group message produces one visible reply containing the exact marker. + - The reply remains scoped to the originating conversation. + docsRefs: + - docs/concepts/qa-e2e-automation.md + - docs/channels/qa-channel.md + codeRefs: + - extensions/qa-lab/src/qa-transport.ts + - extensions/qa-lab/src/suite.ts + execution: + kind: flow + summary: Run the shared channel canary through QA Channel, Crabline, or a live adapter. + transportPolicy: + requireGroupMention: true + config: + conversationId: qa-routing-primary + expectedMarker: QA-CHANNEL-CANARY-OK + +flow: + steps: + - name: delivers one canary reply + actions: + - call: waitForGatewayHealthy + args: [{ ref: env }, 60000] + - call: waitForTransportReady + args: [{ ref: env }, 60000] + - resetTransport: true + - sendInbound: + conversation: + id: { ref: config.conversationId } + kind: group + senderId: driver + senderName: QA Driver + text: + expr: "`@openclaw reply exactly: ${config.expectedMarker}`" + - waitForOutbound: + conversation: + id: { ref: config.conversationId } + kind: group + textIncludes: { ref: config.expectedMarker } + timeoutMs: + expr: liveTurnTimeoutMs(env, 60000) + saveAs: outbound + - set: matchingOutbound + value: + expr: "state.getSnapshot().messages.filter((message) => message.direction === 'outbound' && message.conversation.id === config.conversationId && String(message.text ?? '').includes(config.expectedMarker))" + - assert: + expr: matchingOutbound.length === 1 + message: + expr: "`expected one canary reply, saw ${matchingOutbound.length}; transcript=${formatTransportTranscript(state, { conversationId: config.conversationId })}`" + detailsExpr: outbound.text diff --git a/qa/scenarios/channels/channel-dm-group-routing.yaml b/qa/scenarios/channels/channel-dm-group-routing.yaml new file mode 100644 index 00000000000..56f8cd404c0 --- /dev/null +++ b/qa/scenarios/channels/channel-dm-group-routing.yaml @@ -0,0 +1,78 @@ +title: Channel DM and group routing + +scenario: + id: channel-dm-group-routing + surface: channels + coverage: + primary: + - channels.dm + - channels.group-messages + secondary: + - runtime.delivery + objective: Verify direct and group replies stay on their originating conversation routes. + successCriteria: + - A direct message receives its marker only on the direct conversation. + - A mentioned group message receives its marker only on the group conversation. + docsRefs: + - docs/concepts/qa-e2e-automation.md + - docs/concepts/session.md + codeRefs: + - extensions/qa-lab/src/qa-transport.ts + - src/routing/resolve-route.ts + execution: + kind: flow + summary: Send sequential DM and group turns and assert route separation. + transportPolicy: + requireGroupMention: true + config: + dmConversationId: qa-routing-dm + dmMarker: QA-DM-ROUTING-OK + groupConversationId: qa-routing-group + groupMarker: QA-GROUP-ROUTING-OK + +flow: + steps: + - name: routes direct and group replies independently + actions: + - call: waitForGatewayHealthy + args: [{ ref: env }, 60000] + - call: waitForTransportReady + args: [{ ref: env }, 60000] + - resetTransport: true + - sendInbound: + conversation: + id: { ref: config.dmConversationId } + kind: direct + senderId: driver + senderName: QA Driver + text: + expr: "`Reply exactly: ${config.dmMarker}`" + - waitForOutbound: + conversation: + id: { ref: config.dmConversationId } + kind: direct + textIncludes: { ref: config.dmMarker } + timeoutMs: + expr: liveTurnTimeoutMs(env, 60000) + saveAs: dmOutbound + - sendInbound: + conversation: + id: { ref: config.groupConversationId } + kind: group + senderId: driver + senderName: QA Driver + text: + expr: "`@openclaw reply exactly: ${config.groupMarker}`" + - waitForOutbound: + conversation: + id: { ref: config.groupConversationId } + kind: group + textIncludes: { ref: config.groupMarker } + timeoutMs: + expr: liveTurnTimeoutMs(env, 60000) + saveAs: groupOutbound + - assert: + expr: "dmOutbound.conversation.id !== groupOutbound.conversation.id && !String(dmOutbound.text ?? '').includes(config.groupMarker) && !String(groupOutbound.text ?? '').includes(config.dmMarker)" + message: + expr: "`DM/group routes crossed; transcript=${formatTransportTranscript(state)}`" + detailsExpr: "`${dmOutbound.conversation.kind}:${dmOutbound.text} | ${groupOutbound.conversation.kind}:${groupOutbound.text}`" diff --git a/qa/scenarios/channels/channel-mention-gating.yaml b/qa/scenarios/channels/channel-mention-gating.yaml new file mode 100644 index 00000000000..1ab550c3d06 --- /dev/null +++ b/qa/scenarios/channels/channel-mention-gating.yaml @@ -0,0 +1,69 @@ +title: Channel mention gating + +scenario: + id: channel-mention-gating + surface: channels + coverage: + primary: + - channels.group-messages + secondary: + - runtime.delivery + objective: Verify required-mention group traffic stays quiet until the bot is explicitly mentioned. + successCriteria: + - Unmentioned group traffic produces no outbound reply. + - A subsequent mentioned turn receives exactly one marker reply. + docsRefs: + - docs/channels/group-messages.md + - docs/concepts/qa-e2e-automation.md + codeRefs: + - extensions/qa-channel/src/inbound.ts + - extensions/qa-lab/src/qa-transport.ts + execution: + kind: flow + summary: Prove the shared mention gate with a quiet turn followed by a mentioned turn. + transportPolicy: + requireGroupMention: true + config: + conversationId: qa-routing-mention + expectedMarker: QA-MENTION-GATING-OK + +flow: + steps: + - name: blocks unmentioned traffic and accepts a mention + actions: + - call: waitForGatewayHealthy + args: [{ ref: env }, 60000] + - call: waitForTransportReady + args: [{ ref: env }, 60000] + - resetTransport: true + - set: outboundStartIndex + value: + expr: "state.getSnapshot().messages.filter((message) => message.direction === 'outbound').length" + - sendInbound: + conversation: + id: { ref: config.conversationId } + kind: group + senderId: driver + senderName: QA Driver + text: + expr: "`reply exactly: ${config.expectedMarker}`" + - waitForNoOutbound: + quietMs: 1500 + sinceIndex: { ref: outboundStartIndex } + - sendInbound: + conversation: + id: { ref: config.conversationId } + kind: group + senderId: driver + senderName: QA Driver + text: + expr: "`@openclaw reply exactly: ${config.expectedMarker}`" + - waitForOutbound: + conversation: + id: { ref: config.conversationId } + kind: group + textIncludes: { ref: config.expectedMarker } + timeoutMs: + expr: liveTurnTimeoutMs(env, 60000) + saveAs: outbound + detailsExpr: outbound.text diff --git a/qa/scenarios/channels/channel-multi-actor-ordering.yaml b/qa/scenarios/channels/channel-multi-actor-ordering.yaml new file mode 100644 index 00000000000..637d860f37b --- /dev/null +++ b/qa/scenarios/channels/channel-multi-actor-ordering.yaml @@ -0,0 +1,76 @@ +title: Channel multi-actor ordering + +scenario: + id: channel-multi-actor-ordering + surface: channels + coverage: + primary: + - runtime.turn-ordering + secondary: + - channels.group-messages + objective: Verify blocked secondary-actor traffic does not poison a later allowlisted driver turn. + successCriteria: + - The observer turn remains unanswered. + - The later driver turn receives the exact marker once. + docsRefs: + - docs/concepts/qa-e2e-automation.md + - docs/channels/group-messages.md + codeRefs: + - extensions/qa-lab/src/qa-transport.ts + - extensions/qa-matrix/src/adapter.runtime.ts + execution: + kind: flow + summary: Preserve actor order across a blocked observer turn and a successful driver turn. + transportPolicy: + requireGroupMention: true + senderAllowlist: + - driver + config: + conversationId: qa-routing-ordering + blockedMarker: QA-BLOCKED-ACTOR-SHOULD-NOT-REPLY + expectedMarker: QA-MULTI-ACTOR-ORDERING-OK + +flow: + steps: + - name: preserves ordering after a blocked actor + actions: + - call: waitForGatewayHealthy + args: [{ ref: env }, 60000] + - call: waitForTransportReady + args: [{ ref: env }, 60000] + - resetTransport: true + - set: outboundStartIndex + value: + expr: "state.getSnapshot().messages.filter((message) => message.direction === 'outbound').length" + - sendInbound: + conversation: + id: { ref: config.conversationId } + kind: group + senderId: observer + senderName: QA Observer + text: + expr: "`@openclaw reply exactly: ${config.blockedMarker}`" + - waitForNoOutbound: + quietMs: 1500 + sinceIndex: { ref: outboundStartIndex } + - sendInbound: + conversation: + id: { ref: config.conversationId } + kind: group + senderId: driver + senderName: QA Driver + text: + expr: "`@openclaw reply exactly: ${config.expectedMarker}`" + - waitForOutbound: + conversation: + id: { ref: config.conversationId } + kind: group + textIncludes: { ref: config.expectedMarker } + timeoutMs: + expr: liveTurnTimeoutMs(env, 60000) + saveAs: outbound + - assert: + expr: "!state.getSnapshot().messages.some((message) => message.direction === 'outbound' && String(message.text ?? '').includes(config.blockedMarker))" + message: + expr: "`blocked actor produced a reply; transcript=${formatTransportTranscript(state)}`" + detailsExpr: outbound.text diff --git a/qa/scenarios/channels/channel-secondary-conversation-isolation.yaml b/qa/scenarios/channels/channel-secondary-conversation-isolation.yaml new file mode 100644 index 00000000000..26840498be1 --- /dev/null +++ b/qa/scenarios/channels/channel-secondary-conversation-isolation.yaml @@ -0,0 +1,77 @@ +title: Channel secondary conversation isolation + +scenario: + id: channel-secondary-conversation-isolation + surface: channels + coverage: + primary: + - channels.group-messages + secondary: + - runtime.turn-ordering + objective: Verify a reply in a secondary conversation stays isolated from the primary conversation. + successCriteria: + - The primary conversation receives only its marker. + - The secondary conversation receives only its marker. + docsRefs: + - docs/concepts/session.md + - docs/concepts/qa-e2e-automation.md + codeRefs: + - extensions/qa-lab/src/qa-transport.ts + - extensions/qa-matrix/src/adapter.runtime.ts + execution: + kind: flow + summary: Send sequential turns to primary and secondary channel conversations and assert isolation. + transportPolicy: + requireGroupMention: true + config: + primaryConversationId: primary + primaryMarker: QA-PRIMARY-CONVERSATION-OK + secondaryConversationId: secondary + secondaryMarker: QA-SECONDARY-CONVERSATION-OK + +flow: + steps: + - name: isolates primary and secondary conversations + actions: + - call: waitForGatewayHealthy + args: [{ ref: env }, 60000] + - call: waitForTransportReady + args: [{ ref: env }, 60000] + - resetTransport: true + - sendInbound: + conversation: + id: { ref: config.primaryConversationId } + kind: group + senderId: driver + senderName: QA Driver + text: + expr: "`@openclaw reply exactly: ${config.primaryMarker}`" + - waitForOutbound: + conversation: + id: { ref: config.primaryConversationId } + kind: group + textIncludes: { ref: config.primaryMarker } + timeoutMs: + expr: liveTurnTimeoutMs(env, 60000) + saveAs: primaryOutbound + - sendInbound: + conversation: + id: { ref: config.secondaryConversationId } + kind: group + senderId: driver + senderName: QA Driver + text: + expr: "`@openclaw reply exactly: ${config.secondaryMarker}`" + - waitForOutbound: + conversation: + id: { ref: config.secondaryConversationId } + kind: group + textIncludes: { ref: config.secondaryMarker } + timeoutMs: + expr: liveTurnTimeoutMs(env, 60000) + saveAs: secondaryOutbound + - assert: + expr: "!String(primaryOutbound.text ?? '').includes(config.secondaryMarker) && !String(secondaryOutbound.text ?? '').includes(config.primaryMarker)" + message: + expr: "`secondary conversation leaked into primary route; transcript=${formatTransportTranscript(state)}`" + detailsExpr: "`${primaryOutbound.conversation.id} -> ${secondaryOutbound.conversation.id}`" diff --git a/qa/scenarios/channels/channel-sender-allowlist.yaml b/qa/scenarios/channels/channel-sender-allowlist.yaml new file mode 100644 index 00000000000..4e586f46b31 --- /dev/null +++ b/qa/scenarios/channels/channel-sender-allowlist.yaml @@ -0,0 +1,71 @@ +title: Channel sender allowlist block and override + +scenario: + id: channel-sender-allowlist + surface: channels + coverage: + primary: + - channels.group-messages + secondary: + - runtime.delivery + objective: Verify a blocked secondary actor stays quiet while the configured driver override still triggers a reply. + successCriteria: + - A mentioned message from the observer actor produces no reply. + - A later mentioned message from the allowlisted driver receives the exact marker. + docsRefs: + - docs/channels/group-messages.md + - docs/concepts/qa-e2e-automation.md + codeRefs: + - extensions/qa-lab/src/qa-transport.ts + - src/channels/turn/kernel.ts + execution: + kind: flow + summary: Exercise sender allowlist rejection and the configured driver override in one ordered flow. + transportPolicy: + requireGroupMention: true + senderAllowlist: + - driver + config: + conversationId: qa-routing-allowlist + expectedMarker: QA-ALLOWLIST-OVERRIDE-OK + +flow: + steps: + - name: blocks observer and allows driver + actions: + - call: waitForGatewayHealthy + args: [{ ref: env }, 60000] + - call: waitForTransportReady + args: [{ ref: env }, 60000] + - resetTransport: true + - set: outboundStartIndex + value: + expr: "state.getSnapshot().messages.filter((message) => message.direction === 'outbound').length" + - sendInbound: + conversation: + id: { ref: config.conversationId } + kind: group + senderId: observer + senderName: QA Observer + text: + expr: "`@openclaw reply exactly: ${config.expectedMarker}`" + - waitForNoOutbound: + quietMs: 1500 + sinceIndex: { ref: outboundStartIndex } + - sendInbound: + conversation: + id: { ref: config.conversationId } + kind: group + senderId: driver + senderName: QA Driver + text: + expr: "`@openclaw reply exactly: ${config.expectedMarker}`" + - waitForOutbound: + conversation: + id: { ref: config.conversationId } + kind: group + textIncludes: { ref: config.expectedMarker } + timeoutMs: + expr: liveTurnTimeoutMs(env, 60000) + saveAs: outbound + detailsExpr: outbound.text diff --git a/qa/scenarios/channels/channel-top-level-reply-shape.yaml b/qa/scenarios/channels/channel-top-level-reply-shape.yaml new file mode 100644 index 00000000000..649db6439b5 --- /dev/null +++ b/qa/scenarios/channels/channel-top-level-reply-shape.yaml @@ -0,0 +1,60 @@ +title: Channel top-level reply shape + +scenario: + id: channel-top-level-reply-shape + surface: channels + coverage: + primary: + - channels.group-visible-replies + secondary: + - runtime.delivery + objective: Verify a top-level group prompt produces a top-level reply without thread or quote relation metadata. + successCriteria: + - The reply contains the exact marker. + - The normalized reply has neither threadId nor replyToId. + docsRefs: + - docs/concepts/qa-e2e-automation.md + - docs/channels/group-messages.md + codeRefs: + - extensions/qa-lab/src/qa-transport.ts + - extensions/qa-lab/src/runtime-api.ts + execution: + kind: flow + summary: Assert portable top-level reply shape while native adapters normalize transport relations. + transportPolicy: + requireGroupMention: true + topLevelReplies: true + config: + conversationId: qa-routing-top-level + expectedMarker: QA-TOP-LEVEL-REPLY-OK + +flow: + steps: + - name: keeps the reply top-level + actions: + - call: waitForGatewayHealthy + args: [{ ref: env }, 60000] + - call: waitForTransportReady + args: [{ ref: env }, 60000] + - resetTransport: true + - sendInbound: + conversation: + id: { ref: config.conversationId } + kind: group + senderId: driver + senderName: QA Driver + text: + expr: "`@openclaw reply exactly: ${config.expectedMarker}`" + - waitForOutbound: + conversation: + id: { ref: config.conversationId } + kind: group + textIncludes: { ref: config.expectedMarker } + timeoutMs: + expr: liveTurnTimeoutMs(env, 60000) + saveAs: outbound + - assert: + expr: "!outbound.threadId && (transport.id === 'qa-channel' || !outbound.replyToId)" + message: + expr: "`expected top-level reply, got threadId=${outbound.threadId ?? ''} replyToId=${outbound.replyToId ?? ''}`" + detailsExpr: outbound.text diff --git a/src/plugin-sdk/qa-channel.ts b/src/plugin-sdk/qa-channel.ts index 1cf1de749d0..3e2cb001d77 100644 --- a/src/plugin-sdk/qa-channel.ts +++ b/src/plugin-sdk/qa-channel.ts @@ -18,7 +18,7 @@ import type { export type * from "./qa-channel-protocol.js"; type QaTargetParts = { - chatType: "direct" | "channel"; + chatType: "direct" | "channel" | "group"; conversationId: string; threadId?: string; }; diff --git a/src/plugin-sdk/qa-runner-runtime.ts b/src/plugin-sdk/qa-runner-runtime.ts index a4afcedced7..6b0b2b230ee 100644 --- a/src/plugin-sdk/qa-runner-runtime.ts +++ b/src/plugin-sdk/qa-runner-runtime.ts @@ -15,11 +15,18 @@ import type { QaBusOutboundMessageInput, } from "./qa-channel-protocol.js"; +type QaRunnerTransportPolicy = { + requireGroupMention?: true; + senderAllowlist?: readonly string[]; + topLevelReplies?: true; +}; + type QaRunnerAdapterOptions = { repoRoot?: string; sutAccountId?: string; credentialSource?: string; credentialRole?: string; + transportPolicy?: QaRunnerTransportPolicy; }; type QaRunnerMessageRecorder = {