From 04601ae81a6d395966ea440dafa2b81dd0f9db36 Mon Sep 17 00:00:00 2001 From: Dallin Romney Date: Fri, 3 Jul 2026 15:54:27 -0700 Subject: [PATCH] refactor(qa): simplify transport adapter contracts (#99632) * feat(qa): add transport capability contracts * fix(qa): scope sequence support by channel * refactor(qa): simplify transport adapter contracts * fix(qa): clarify unsupported transport method errors --- .../qa-lab/src/crabline-transport.test.ts | 8 +- extensions/qa-lab/src/crabline-transport.ts | 41 ++-- extensions/qa-lab/src/manual-lane.runtime.ts | 16 +- .../qa-lab/src/qa-channel-transport.test.ts | 20 +- extensions/qa-lab/src/qa-channel-transport.ts | 14 +- .../qa-lab/src/qa-transport-registry.test.ts | 89 ++++++++- .../qa-lab/src/qa-transport-registry.ts | 112 +++++++++-- extensions/qa-lab/src/qa-transport.ts | 66 +------ .../qa-lab/src/scenario-flow-runner.test.ts | 24 ++- extensions/qa-lab/src/scenario-flow-runner.ts | 6 + .../qa-lab/src/scenario-runtime-api.test.ts | 21 +- extensions/qa-lab/src/scenario-runtime-api.ts | 30 +-- extensions/qa-lab/src/self-check.ts | 8 +- .../qa-lab/src/suite-runtime-flow.test.ts | 13 +- extensions/qa-lab/src/suite.ts | 182 ++++++++++-------- 15 files changed, 401 insertions(+), 249 deletions(-) diff --git a/extensions/qa-lab/src/crabline-transport.test.ts b/extensions/qa-lab/src/crabline-transport.test.ts index 37daa671af7b..59b2258657a7 100644 --- a/extensions/qa-lab/src/crabline-transport.test.ts +++ b/extensions/qa-lab/src/crabline-transport.test.ts @@ -71,7 +71,8 @@ describe("crabline transport", () => { }); try { - await transport.sendNativeCommand({ + expect(transport.sendNativeCommand).toBeTypeOf("function"); + await transport.sendNativeCommand?.({ command: "stop", conversation: { id: "alice", kind: "direct" }, senderId: "alice", @@ -218,8 +219,9 @@ describe("crabline transport", () => { text: "final marker", }); + expect(transport.waitForOutboundSequence).toBeTypeOf("function"); await expect( - transport.waitForOutboundSequence({ + transport.waitForOutboundSequence!({ conversationId: "-1001234567890", finalSettleMs: 0, finalTextIncludes: "final marker", @@ -248,6 +250,8 @@ describe("crabline transport", () => { try { expect(transport.id).toBe("crabline"); expect(transport.requiredPluginIds).toEqual(["slack"]); + expect(transport.sendNativeCommand).toBeUndefined(); + expect(transport.waitForOutboundSequence).toBeUndefined(); expect(transport.createGatewayConfig({ baseUrl: "http://127.0.0.1:1" })).toMatchObject({ channels: { slack: { diff --git a/extensions/qa-lab/src/crabline-transport.ts b/extensions/qa-lab/src/crabline-transport.ts index 517ef95dfdcc..38466f873a0b 100644 --- a/extensions/qa-lab/src/crabline-transport.ts +++ b/extensions/qa-lab/src/crabline-transport.ts @@ -338,6 +338,11 @@ class QaCrablineTransport extends QaStateBackedTransportAdapter { readonly #adapter: StartedOpenClawCrablineAdapter; readonly #selection: OpenClawCrablineChannelDriverSelection; readonly #state: QaCrablineTransportState; + readonly sendNativeCommand?: (input: QaTransportNativeCommandInput) => Promise; + readonly waitForOutboundSequence?: (input: QaTransportOutboundSequenceMatch) => Promise<{ + events: QaTransportOutboundEvent[]; + final: QaBusMessage; + }>; constructor(params: { adapter: StartedOpenClawCrablineAdapter; @@ -354,6 +359,21 @@ class QaCrablineTransport extends QaStateBackedTransportAdapter { this.#adapter = params.adapter; this.#selection = params.selection; this.#state = params.state; + if (params.selection.channel === "telegram") { + this.sendNativeCommand = async (input) => { + const { command, ...message } = input; + await this.sendInbound({ + ...message, + text: `/${command}`, + nativeCommand: { name: command }, + }); + }; + this.waitForOutboundSequence = async (input) => + await waitForQaTransportOutboundSequence({ + input, + readEvents: () => this.#state.getOutboundEvents(), + }); + } } createGatewayConfig = (params: { baseUrl: string }): QaTransportGatewayConfig => @@ -378,27 +398,6 @@ class QaCrablineTransport extends QaStateBackedTransportAdapter { createRuntimeEnvPatch = () => this.#adapter.createChannelDriverSmokeEnv({}); - override async sendNativeCommand(input: QaTransportNativeCommandInput): Promise { - if (this.#selection.channel !== "telegram") { - throw new Error( - `Crabline ${this.#selection.channel} does not support native command injection.`, - ); - } - const { command, ...message } = input; - await this.sendInbound({ - ...message, - text: `/${command}`, - nativeCommand: { name: command }, - }); - } - - override async waitForOutboundSequence(input: QaTransportOutboundSequenceMatch) { - return await waitForQaTransportOutboundSequence({ - input, - readEvents: () => this.#state.getOutboundEvents(), - }); - } - handleAction = async (_params: { action: QaTransportActionName; args: Record; diff --git a/extensions/qa-lab/src/manual-lane.runtime.ts b/extensions/qa-lab/src/manual-lane.runtime.ts index b5b6a9ef84a3..5aa2bb9b5fb4 100644 --- a/extensions/qa-lab/src/manual-lane.runtime.ts +++ b/extensions/qa-lab/src/manual-lane.runtime.ts @@ -77,6 +77,7 @@ export async function runQaManualLane(params: QaManualLaneParams) { let gateway: Awaited> | undefined; let lab: Awaited> | undefined; let mock: Awaited> | undefined; + let transportCleanup: (() => Promise) | undefined; let result: ManualLaneResult | undefined; let cleanupError: Error | undefined; let runError: unknown; @@ -86,10 +87,14 @@ export async function runQaManualLane(params: QaManualLaneParams) { repoRoot: params.repoRoot, embeddedGateway: "disabled", }); - const transport = createQaTransportAdapter({ - id: params.transportId ?? "qa-channel", + const transportFactoryResult = await createQaTransportAdapter({ + channelId: params.transportId ?? "qa-channel", + driver: params.transportId ?? "qa-channel", + outputDir: params.repoRoot, state: lab.state, }); + const transport = transportFactoryResult.adapter; + transportCleanup = transportFactoryResult.cleanup; mock = await startQaProviderServer(params.providerMode); gateway = await startQaGatewayChild({ repoRoot: params.repoRoot, @@ -164,7 +169,12 @@ export async function runQaManualLane(params: QaManualLaneParams) { } catch (error) { runError = error; } finally { - cleanupError = await stopManualLaneResources({ gateway, lab, mock }); + let transportCleanupError: Error | undefined; + await transportCleanup?.().catch((error: unknown) => { + transportCleanupError = normalizeManualLaneCleanupError(error); + }); + const resourceCleanupError = await stopManualLaneResources({ gateway, lab, mock }); + cleanupError = transportCleanupError ?? resourceCleanupError; } if (runError) { throw new Error(formatErrorMessage(runError), { cause: runError }); diff --git a/extensions/qa-lab/src/qa-channel-transport.test.ts b/extensions/qa-lab/src/qa-channel-transport.test.ts index 0ddb4aaa5490..974c51fbf4ab 100644 --- a/extensions/qa-lab/src/qa-channel-transport.test.ts +++ b/extensions/qa-lab/src/qa-channel-transport.test.ts @@ -98,18 +98,18 @@ describe("qa channel transport", () => { ).rejects.toThrow("last probe error: channels.status exploded"); }); - it("inherits the shared normalized message capabilities", async () => { + it("uses the shared normalized message state", async () => { const transport = createQaChannelTransport(createQaBusState()); - const inbound = await transport.capabilities.sendInboundMessage({ + const inbound = await transport.sendInbound({ accountId: "default", conversation: { id: "dm:qa-operator", kind: "direct" }, senderId: "qa-operator", text: "hello from the operator", }); - expect(transport.capabilities.getNormalizedMessageState().messages).toHaveLength(1); - const message = await transport.capabilities.readNormalizedMessage({ + expect(transport.state.getSnapshot().messages).toHaveLength(1); + const message = await transport.state.readMessage({ messageId: inbound.id, }); if (!message) { @@ -161,11 +161,11 @@ describe("qa channel transport", () => { let injected = false; await expect( - transport.capabilities.waitForCondition( + transport.waitForCondition( async () => { if (!injected) { injected = true; - await transport.capabilities.injectOutboundMessage({ + await transport.state.addOutboundMessage({ accountId: "default", to: "dm:qa-operator", text: "⚠️ agent failed before reply: synthetic failure for wait helper", @@ -182,22 +182,20 @@ describe("qa channel transport", () => { it("captures a fresh failure cursor for each wait helper call", async () => { const transport = createQaChannelTransport(createQaBusState()); - await transport.capabilities.injectOutboundMessage({ + await transport.state.addOutboundMessage({ accountId: "default", to: "dm:qa-operator", text: "⚠️ agent failed before reply: stale failure should not leak", }); - await expect(transport.capabilities.waitForCondition(async () => "ok", 50, 10)).resolves.toBe( - "ok", - ); + await expect(transport.waitForCondition(async () => "ok", 50, 10)).resolves.toBe("ok"); }); it("keeps oversized wait helper intervals within the timeout", async () => { const transport = createQaChannelTransport(createQaBusState()); await expect( - transport.capabilities.waitForCondition(async () => undefined, 5, Number.MAX_SAFE_INTEGER), + transport.waitForCondition(async () => undefined, 5, Number.MAX_SAFE_INTEGER), ).rejects.toThrow("timed out after 5ms"); }); }); diff --git a/extensions/qa-lab/src/qa-channel-transport.ts b/extensions/qa-lab/src/qa-channel-transport.ts index cae2ae18ff91..292eefc084bf 100644 --- a/extensions/qa-lab/src/qa-channel-transport.ts +++ b/extensions/qa-lab/src/qa-channel-transport.ts @@ -5,12 +5,16 @@ import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; import type { QaBusState } from "./bus-state.js"; import { QaSuiteInfraError } from "./errors.js"; import { getQaProvider } from "./providers/index.js"; -import { QaStateBackedTransportAdapter } from "./qa-transport.js"; +import { + QaStateBackedTransportAdapter, + waitForQaTransportOutboundSequence, +} from "./qa-transport.js"; import type { QaTransportActionName, QaTransportGatewayConfig, QaTransportGatewayClient, QaTransportNativeCommandInput, + QaTransportOutboundSequenceMatch, QaTransportReportParams, } from "./qa-transport.js"; import { qaChannelPlugin } from "./runtime-api.js"; @@ -148,7 +152,7 @@ class QaChannelTransport extends QaStateBackedTransportAdapter { replyChannel: QA_CHANNEL_ID, replyTo: target, }); - override async sendNativeCommand(input: QaTransportNativeCommandInput): Promise { + async sendNativeCommand(input: QaTransportNativeCommandInput): Promise { const { command, ...message } = input; await this.sendInbound({ ...message, @@ -156,6 +160,12 @@ class QaChannelTransport extends QaStateBackedTransportAdapter { nativeCommand: { name: command }, }); } + async waitForOutboundSequence(input: QaTransportOutboundSequenceMatch) { + return await waitForQaTransportOutboundSequence({ + input, + readEvents: () => this.state.getSnapshot().events, + }); + } handleAction = handleQaChannelAction; createReportNotes = createQaChannelReportNotes; } diff --git a/extensions/qa-lab/src/qa-transport-registry.test.ts b/extensions/qa-lab/src/qa-transport-registry.test.ts index abe9dd4ce327..79dc5f7be1e9 100644 --- a/extensions/qa-lab/src/qa-transport-registry.test.ts +++ b/extensions/qa-lab/src/qa-transport-registry.test.ts @@ -1,6 +1,27 @@ // Qa Lab tests cover qa transport registry plugin behavior. -import { describe, expect, it } from "vitest"; -import { normalizeQaTransportId } from "./qa-transport-registry.js"; +import { describe, expect, it, vi } from "vitest"; +import { createQaBusState } from "./bus-state.js"; +import { createQaChannelTransport } from "./qa-channel-transport.js"; +import { + createQaTransportAdapter, + createQaTransportAdapterFactoryRegistry, + normalizeQaTransportId, + type QaTransportAdapterFactory, + type QaTransportFactoryContext, +} from "./qa-transport-registry.js"; +import type { QaTransportAdapter } from "./qa-transport.js"; + +function createFactoryContext( + overrides: Partial = {}, +): QaTransportFactoryContext { + return { + channelId: "qa-channel", + driver: "qa-channel", + outputDir: ".artifacts/qa-e2e/transport-contract-test", + state: createQaBusState(), + ...overrides, + }; +} describe("qa transport registry", () => { it("rejects inherited prototype keys as unsupported transport ids", () => { @@ -9,4 +30,68 @@ describe("qa transport registry", () => { "unsupported QA transport: __proto__", ); }); + + it("creates QA Channel through the default async registry", async () => { + const created = await createQaTransportAdapter(createFactoryContext()); + + expect(created.adapter.id).toBe("qa-channel"); + await created.cleanup(); + }); + + it("selects an injected matching factory", async () => { + const adapter = createQaChannelTransport(createQaBusState()); + const skippedCreate = vi.fn(async () => adapter); + const selectedCreate = vi.fn(async () => adapter); + const factories: QaTransportAdapterFactory[] = [ + { id: "skipped", matches: () => false, create: skippedCreate }, + { id: "selected", matches: () => true, create: selectedCreate }, + ]; + const registry = createQaTransportAdapterFactoryRegistry(factories); + + const created = await registry.create(createFactoryContext()); + + expect(created.adapter).toBe(adapter); + expect(skippedCreate).not.toHaveBeenCalled(); + expect(selectedCreate).toHaveBeenCalledOnce(); + }); + + it("returns cleanup owned by the selected adapter", async () => { + const cleanup = vi.fn(async () => undefined); + const adapter: QaTransportAdapter = createQaChannelTransport(createQaBusState()); + adapter.cleanup = cleanup; + const factory: QaTransportAdapterFactory = { + id: "cleanup", + matches: () => true, + async create() { + return adapter; + }, + }; + const registry = createQaTransportAdapterFactoryRegistry([factory]); + const created = await registry.create(createFactoryContext()); + + await created.cleanup(); + + expect(cleanup).toHaveBeenCalledOnce(); + }); + + it("reports no-match and startup failures with transport context", async () => { + const context = createFactoryContext(); + const emptyRegistry = createQaTransportAdapterFactoryRegistry([]); + await expect(emptyRegistry.create(context)).rejects.toThrow( + "no QA transport factory for qa-channel:qa-channel", + ); + + const brokenRegistry = createQaTransportAdapterFactoryRegistry([ + { + id: "broken", + matches: () => true, + async create() { + throw new Error("provider boot failed"); + }, + }, + ]); + await expect(brokenRegistry.create(context)).rejects.toThrow( + "broken failed to create QA transport qa-channel:qa-channel: provider boot failed", + ); + }); }); diff --git a/extensions/qa-lab/src/qa-transport-registry.ts b/extensions/qa-lab/src/qa-transport-registry.ts index 7e084f96f8b8..7ceadf647d44 100644 --- a/extensions/qa-lab/src/qa-transport-registry.ts +++ b/extensions/qa-lab/src/qa-transport-registry.ts @@ -7,37 +7,113 @@ import { import type { QaTransportAdapter } from "./qa-transport.js"; export type QaTransportId = "qa-channel"; +export type QaTransportDriver = QaTransportId | "crabline"; + +export type QaTransportFactoryContext = { + channelId: string; + driver: QaTransportDriver; + outputDir: string; + state: QaBusState; +}; + +export type QaTransportAdapterFactoryResult = { + adapter: QaTransportAdapter; + cleanup: () => Promise; +}; + +export type QaTransportAdapterFactory = { + id: string; + matches: (context: Pick) => boolean; + create: (context: QaTransportFactoryContext) => Promise; +}; + +export type QaTransportAdapterFactoryRegistry = { + create: (context: QaTransportFactoryContext) => Promise; +}; const DEFAULT_QA_TRANSPORT_ID: QaTransportId = "qa-channel"; -const QA_TRANSPORT_REGISTRY = { - "qa-channel": { - create: createQaChannelTransport, - defaultSuiteConcurrency: QA_CHANNEL_DEFAULT_SUITE_CONCURRENCY, +const QA_CHANNEL_TRANSPORT_FACTORY: QaTransportAdapterFactory = { + id: "qa-channel", + matches: ({ channelId, driver }) => driver === "qa-channel" && channelId === "qa-channel", + async create(context) { + return createQaChannelTransport(context.state); }, -} as const satisfies Record< - QaTransportId, - { - create: (state: QaBusState) => QaTransportAdapter; - defaultSuiteConcurrency: number; +}; + +const CRABLINE_TRANSPORT_FACTORY: QaTransportAdapterFactory = { + id: "crabline", + matches: ({ driver }) => driver === "crabline", + async create(context) { + const { resolveOpenClawCrablineChannelDriverSelection } = await import("@openclaw/crabline"); + const selection = resolveOpenClawCrablineChannelDriverSelection({ channel: context.channelId }); + const { createQaCrablineTransportAdapter } = await import("./crabline-transport.js"); + return await createQaCrablineTransportAdapter({ + outputDir: context.outputDir, + selection, + state: context.state, + }); + }, +}; + +const DEFAULT_QA_TRANSPORT_FACTORIES = [ + QA_CHANNEL_TRANSPORT_FACTORY, + CRABLINE_TRANSPORT_FACTORY, +] as const; + +function requireQaTransportFactory( + factories: readonly QaTransportAdapterFactory[], + context: Pick, +) { + const factory = factories.find((candidate) => candidate.matches(context)); + if (!factory) { + throw new Error(`no QA transport factory for ${context.driver}:${context.channelId}`); } ->; + return factory; +} + +export function createQaTransportAdapterFactoryRegistry( + factories: readonly QaTransportAdapterFactory[] = DEFAULT_QA_TRANSPORT_FACTORIES, +): QaTransportAdapterFactoryRegistry { + return { + async create(context) { + const factory = requireQaTransportFactory(factories, context); + let adapter: QaTransportAdapter; + try { + adapter = await factory.create(context); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + throw new Error( + `${factory.id} failed to create QA transport ${context.driver}:${context.channelId}: ${message}`, + { cause: error }, + ); + } + return { + adapter, + cleanup: async () => { + await adapter.cleanup?.(); + }, + }; + }, + }; +} + +const qaTransportAdapterFactoryRegistry = createQaTransportAdapterFactoryRegistry(); export function normalizeQaTransportId(input?: string | null): QaTransportId { const transportId = input?.trim() || DEFAULT_QA_TRANSPORT_ID; - if (Object.hasOwn(QA_TRANSPORT_REGISTRY, transportId)) { - return transportId as QaTransportId; + if (transportId === "qa-channel") { + return transportId; } throw new Error(`unsupported QA transport: ${transportId}`); } -export function createQaTransportAdapter(params: { - id: QaTransportId; - state: QaBusState; -}): QaTransportAdapter { - return QA_TRANSPORT_REGISTRY[params.id].create(params.state); +export async function createQaTransportAdapter( + context: QaTransportFactoryContext, +): Promise { + return await qaTransportAdapterFactoryRegistry.create(context); } export function defaultQaSuiteConcurrencyForTransport(id: QaTransportId): number { - return QA_TRANSPORT_REGISTRY[id].defaultSuiteConcurrency; + return id === "qa-channel" ? QA_CHANNEL_DEFAULT_SUITE_CONCURRENCY : 1; } diff --git a/extensions/qa-lab/src/qa-transport.ts b/extensions/qa-lab/src/qa-transport.ts index a9b277d9002c..d732bbfb7de8 100644 --- a/extensions/qa-lab/src/qa-transport.ts +++ b/extensions/qa-lab/src/qa-transport.ts @@ -99,32 +99,6 @@ export type QaTransportNativeCommandInput = Omit< command: string; }; -export type QaTransportCapabilities = { - sendInboundMessage: QaTransportState["addInboundMessage"]; - injectOutboundMessage: QaTransportState["addOutboundMessage"]; - waitForOutboundMessage: (input: QaBusWaitForInput) => Promise; - getNormalizedMessageState: () => QaBusStateSnapshot; - resetNormalizedMessageState: () => Promise; - readNormalizedMessage: QaTransportState["readMessage"]; - executeGenericAction: (params: { - action: QaTransportActionName; - args: Record; - cfg: OpenClawConfig; - accountId?: string | null; - }) => Promise; - waitForReady: (params: { - gateway: QaTransportGatewayClient; - timeoutMs?: number; - pollIntervalMs?: number; - }) => Promise; - waitForCondition: ( - check: () => T | Promise | null | undefined, - timeoutMs?: number, - intervalMs?: number, - ) => Promise; - assertNoFailureReplies: (options?: QaTransportFailureAssertionOptions) => void; -}; - export async function waitForQaTransportCondition( check: () => T | Promise | null | undefined, timeoutMs = 15_000, @@ -207,15 +181,19 @@ export type QaTransportAdapter = { requiredPluginIds: readonly string[]; supportedActions: readonly QaTransportActionName[]; state: QaTransportState; - capabilities: QaTransportCapabilities; reset: () => Promise; sendInbound: (input: QaBusInboundMessageInput) => Promise; - sendNativeCommand: (input: QaTransportNativeCommandInput) => Promise; + sendNativeCommand?: (input: QaTransportNativeCommandInput) => Promise; waitForNoOutbound: (input?: QaTransportWaitForNoOutboundInput) => Promise; waitForOutbound: (input: QaTransportOutboundMatch) => Promise; - waitForOutboundSequence: ( + waitForOutboundSequence?: ( input: QaTransportOutboundSequenceMatch, ) => Promise; + waitForCondition: ( + check: () => T | Promise | null | undefined, + timeoutMs?: number, + intervalMs?: number, + ) => Promise; createGatewayConfig: (params: { baseUrl: string }) => QaTransportGatewayConfig; waitReady: (params: { gateway: QaTransportGatewayClient; @@ -246,7 +224,7 @@ export abstract class QaStateBackedTransportAdapter implements QaTransportAdapte readonly requiredPluginIds: readonly string[]; readonly supportedActions: readonly QaTransportActionName[]; readonly state: QaTransportState; - readonly capabilities: QaTransportCapabilities; + readonly waitForCondition: QaTransportAdapter["waitForCondition"]; protected constructor(params: { id: string; @@ -262,22 +240,7 @@ export abstract class QaStateBackedTransportAdapter implements QaTransportAdapte this.requiredPluginIds = params.requiredPluginIds; this.supportedActions = params.supportedActions ?? []; this.state = params.state; - this.capabilities = { - sendInboundMessage: this.state.addInboundMessage.bind(this.state), - injectOutboundMessage: this.state.addOutboundMessage.bind(this.state), - waitForOutboundMessage: this.state.waitFor.bind(this.state), - getNormalizedMessageState: this.state.getSnapshot.bind(this.state), - resetNormalizedMessageState: async () => { - await this.state.reset(); - }, - readNormalizedMessage: this.state.readMessage.bind(this.state), - executeGenericAction: (paramsValue) => this.handleAction(paramsValue), - waitForReady: (paramsLocal) => this.waitReady(paramsLocal), - waitForCondition: createFailureAwareTransportWaitForCondition(this.state), - assertNoFailureReplies: (options) => { - assertNoFailureReplies(this.state, options); - }, - }; + this.waitForCondition = createFailureAwareTransportWaitForCondition(this.state); } abstract createGatewayConfig: (params: { baseUrl: string }) => QaTransportGatewayConfig; @@ -308,10 +271,6 @@ export abstract class QaStateBackedTransportAdapter implements QaTransportAdapte return await this.state.addInboundMessage(input); } - async sendNativeCommand(_input: QaTransportNativeCommandInput): Promise { - throw new Error(`${this.label} does not support native commands.`); - } - async waitForNoOutbound(input: QaTransportWaitForNoOutboundInput = {}) { const quietMs = resolveTimerTimeoutMs(input.quietMs, 1_200, 0); await sleep(quietMs); @@ -350,13 +309,6 @@ export abstract class QaStateBackedTransportAdapter implements QaTransportAdapte }, input.timeoutMs); } - async waitForOutboundSequence(input: QaTransportOutboundSequenceMatch) { - return await waitForQaTransportOutboundSequence({ - input, - readEvents: () => this.state.getSnapshot().events, - }); - } - private outboundSince(sinceIndex = 0) { return this.state .getSnapshot() diff --git a/extensions/qa-lab/src/scenario-flow-runner.test.ts b/extensions/qa-lab/src/scenario-flow-runner.test.ts index 64a0e7c1bc33..d2f2bfa1dd4c 100644 --- a/extensions/qa-lab/src/scenario-flow-runner.test.ts +++ b/extensions/qa-lab/src/scenario-flow-runner.test.ts @@ -19,6 +19,7 @@ function formatTestTranscript(state: ReturnType) { async function runLoadedScenarioFlow( scenarioId: string, params: { + omitOutboundSequence?: boolean; onWaitForOutboundMessage?: (params: { waitCount: number; state: ReturnType; @@ -74,12 +75,16 @@ async function runLoadedScenarioFlow( } throw new Error(`timed out after ${input.timeoutMs}ms waiting for outbound marker`); }, - waitForOutboundSequence: async () => { - throw new Error("outbound sequence not configured for this fixture"); - }, + ...(params.omitOutboundSequence + ? {} + : { + waitForOutboundSequence: async () => { + throw new Error("outbound sequence not configured for this fixture"); + }, + }), }; const api = { - env: {}, + env: { providerMode: "mock-openai" }, transport, state, scenario, @@ -87,6 +92,7 @@ async function runLoadedScenarioFlow( randomUUID: () => "00000000-0000-4000-8000-000000000000", liveTurnTimeoutMs: (_env: unknown, timeoutMs: number) => timeoutMs, waitForGatewayHealthy: async () => undefined, + waitForTransportReady: async () => undefined, waitForQaChannelReady: async () => undefined, waitForNoOutbound: async () => undefined, sleep: async () => undefined, @@ -141,6 +147,16 @@ async function runLoadedScenarioFlow( } describe("scenario-flow-runner", () => { + it("fails when a flow calls a transport method the adapter does not implement", async () => { + await expect( + runLoadedScenarioFlow("channel-message-flows", { + omitOutboundSequence: true, + }), + ).rejects.toThrow( + 'QA scenario "channel-message-flows" cannot run "waitForOutboundSequence": the active transport adapter does not implement this method.', + ); + }); + it("supports qaImport inside flow expressions", async () => { const result = await runScenarioFlow({ api: { diff --git a/extensions/qa-lab/src/scenario-flow-runner.ts b/extensions/qa-lab/src/scenario-flow-runner.ts index fdbc434518de..874000ba10ce 100644 --- a/extensions/qa-lab/src/scenario-flow-runner.ts +++ b/extensions/qa-lab/src/scenario-flow-runner.ts @@ -141,6 +141,12 @@ async function resolveValue(node: unknown, api: QaFlowApi, vars: QaFlowVars): Pr function resolveCallable(path: string, api: QaFlowApi, vars: QaFlowVars) { const { parent, value } = getPathWithParent(createEvalContext(api, vars), path); if (typeof value !== "function") { + if (path.startsWith("transport.")) { + const method = path.slice("transport.".length); + throw new Error( + `QA scenario "${api.scenario.id}" cannot run "${method}": the active transport adapter does not implement this method.`, + ); + } throw new Error(`qa flow callable not found: ${path}`); } return parent ? value.bind(parent) : value; diff --git a/extensions/qa-lab/src/scenario-runtime-api.test.ts b/extensions/qa-lab/src/scenario-runtime-api.test.ts index 90cb45b69c6a..a32b54d6d19f 100644 --- a/extensions/qa-lab/src/scenario-runtime-api.test.ts +++ b/extensions/qa-lab/src/scenario-runtime-api.test.ts @@ -4,7 +4,7 @@ import * as fs from "node:fs/promises"; import path from "node:path"; import { describe, expect, it, vi } from "vitest"; import { createQaBusState } from "./bus-state.js"; -import type { QaTransportCapabilities } from "./qa-transport.js"; +import type { QaTransportAdapter } from "./qa-transport.js"; import { createQaScenarioRuntimeApi, type QaScenarioRuntimeConstants, @@ -113,13 +113,13 @@ const browserAndWebRuntimeTools = [ ] as const; describe("createQaScenarioRuntimeApi", () => { - it("builds a markdown-flow runtime surface from generic transport capabilities", async () => { + it("builds a markdown-flow runtime surface from the transport adapter", async () => { const state = createQaBusState(); const resetSpy = vi.spyOn(state, "reset"); const inboundSpy = vi.spyOn(state, "addInboundMessage"); const outboundSpy = vi.spyOn(state, "addOutboundMessage"); const readSpy = vi.spyOn(state, "readMessage"); - const waitForCondition: QaTransportCapabilities["waitForCondition"] = async ( + const waitForCondition: QaTransportAdapter["waitForCondition"] = async ( check: () => T | Promise | null | undefined, ): Promise => { const value = await check(); @@ -157,20 +157,7 @@ describe("createQaScenarioRuntimeApi", () => { waitForOutboundSequence: vi.fn(async () => { throw new Error("not used"); }), - capabilities: { - waitForCondition, - getNormalizedMessageState: state.getSnapshot.bind(state), - resetNormalizedMessageState: async () => { - state.reset(); - }, - sendInboundMessage: state.addInboundMessage.bind(state), - injectOutboundMessage: state.addOutboundMessage.bind(state), - waitForOutboundMessage: state.waitFor.bind(state), - readNormalizedMessage: state.readMessage.bind(state), - executeGenericAction: vi.fn(async () => undefined), - waitForReady: vi.fn(async () => undefined), - assertNoFailureReplies: vi.fn(), - }, + waitForCondition, }, }; const scenario = { diff --git a/extensions/qa-lab/src/scenario-runtime-api.ts b/extensions/qa-lab/src/scenario-runtime-api.ts index d6025b6cb156..639ea21d8a9d 100644 --- a/extensions/qa-lab/src/scenario-runtime-api.ts +++ b/extensions/qa-lab/src/scenario-runtime-api.ts @@ -8,13 +8,13 @@ type QaScenarioRuntimeFunction = (...args: never[]) => unknown; type QaScenarioTransport = Pick< QaTransportAdapter, - | "capabilities" | "reset" | "sendInbound" | "sendNativeCommand" | "state" | "waitForNoOutbound" | "waitForOutbound" + | "waitForCondition" >; export type QaScenarioRuntimeEnv< @@ -124,7 +124,7 @@ type QaScenarioRuntimeApi< sleep: (ms?: number) => Promise; randomUUID: () => string; runScenario: TDeps["runScenario"]; - waitForCondition: TEnv["transport"]["capabilities"]["waitForCondition"]; + waitForCondition: TEnv["transport"]["waitForCondition"]; waitForOutboundMessage: TDeps["waitForOutboundMessage"]; waitForTransportOutboundMessage: TDeps["waitForTransportOutboundMessage"]; waitForChannelOutboundMessage: TDeps["waitForChannelOutboundMessage"]; @@ -199,11 +199,11 @@ type QaScenarioRuntimeApi< imageUnderstandingPngBase64: string; imageUnderstandingLargePngBase64: string; imageUnderstandingValidPngBase64: string; - getTransportSnapshot: TEnv["transport"]["capabilities"]["getNormalizedMessageState"]; + getTransportSnapshot: TEnv["transport"]["state"]["getSnapshot"]; resetTransport: () => Promise; - injectInboundMessage: TEnv["transport"]["capabilities"]["sendInboundMessage"]; - injectOutboundMessage: TEnv["transport"]["capabilities"]["injectOutboundMessage"]; - readTransportMessage: TEnv["transport"]["capabilities"]["readNormalizedMessage"]; + injectInboundMessage: TEnv["transport"]["state"]["addInboundMessage"]; + injectOutboundMessage: TEnv["transport"]["state"]["addOutboundMessage"]; + readTransportMessage: TEnv["transport"]["state"]["readMessage"]; resetBus: () => Promise; reset: () => Promise; }; @@ -217,16 +217,18 @@ export function createQaScenarioRuntimeApi< deps: TDeps; constants: QaScenarioRuntimeConstants; }): QaScenarioRuntimeApi { + const transport = params.env.transport; + const transportState = transport.state; const resetTransportState = async () => { - await params.env.transport.capabilities.resetNormalizedMessageState(); + await transport.reset(); await params.deps.sleep(100); }; return { env: params.env, lab: params.env.lab, - transport: params.env.transport, - state: params.env.transport.state, + transport, + state: transportState, scenario: params.scenario, config: params.scenario.execution.config ?? {}, fs: params.deps.fs, @@ -234,7 +236,7 @@ export function createQaScenarioRuntimeApi< sleep: params.deps.sleep, randomUUID: params.deps.randomUUID, runScenario: params.deps.runScenario, - waitForCondition: params.env.transport.capabilities.waitForCondition, + waitForCondition: transport.waitForCondition, waitForOutboundMessage: params.deps.waitForOutboundMessage, waitForTransportOutboundMessage: params.deps.waitForTransportOutboundMessage, waitForChannelOutboundMessage: params.deps.waitForChannelOutboundMessage, @@ -309,11 +311,11 @@ export function createQaScenarioRuntimeApi< imageUnderstandingPngBase64: params.constants.imageUnderstandingPngBase64, imageUnderstandingLargePngBase64: params.constants.imageUnderstandingLargePngBase64, imageUnderstandingValidPngBase64: params.constants.imageUnderstandingValidPngBase64, - getTransportSnapshot: params.env.transport.capabilities.getNormalizedMessageState, + getTransportSnapshot: transportState.getSnapshot.bind(transportState), resetTransport: resetTransportState, - injectInboundMessage: params.env.transport.capabilities.sendInboundMessage, - injectOutboundMessage: params.env.transport.capabilities.injectOutboundMessage, - readTransportMessage: params.env.transport.capabilities.readNormalizedMessage, + injectInboundMessage: transportState.addInboundMessage.bind(transportState), + injectOutboundMessage: transportState.addOutboundMessage.bind(transportState), + readTransportMessage: transportState.readMessage.bind(transportState), resetBus: resetTransportState, reset: resetTransportState, }; diff --git a/extensions/qa-lab/src/self-check.ts b/extensions/qa-lab/src/self-check.ts index d9b54f041c6b..f46f69fdc35b 100644 --- a/extensions/qa-lab/src/self-check.ts +++ b/extensions/qa-lab/src/self-check.ts @@ -41,10 +41,13 @@ export async function runQaSelfCheckAgainstState(params: { waitTimeoutMs?: number; }): Promise { const startedAt = new Date(); - const transport = createQaTransportAdapter({ - id: params.transportId ?? "qa-channel", + const transportFactoryResult = await createQaTransportAdapter({ + channelId: params.transportId ?? "qa-channel", + driver: params.transportId ?? "qa-channel", + outputDir: path.dirname(resolveQaSelfCheckOutputPath(params)), state: params.state, }); + const transport = transportFactoryResult.adapter; params.state.reset(); const scenarioResult = await runQaScenario( createQaSelfCheckScenario({ waitTimeoutMs: params.waitTimeoutMs }), @@ -104,6 +107,7 @@ export async function runQaSelfCheckAgainstState(params: { }); await fs.mkdir(path.dirname(outputPath), { recursive: true }); await fs.writeFile(outputPath, report, "utf8"); + await transportFactoryResult.cleanup(); return { outputPath, diff --git a/extensions/qa-lab/src/suite-runtime-flow.test.ts b/extensions/qa-lab/src/suite-runtime-flow.test.ts index 636a5fff93d3..a062550a3041 100644 --- a/extensions/qa-lab/src/suite-runtime-flow.test.ts +++ b/extensions/qa-lab/src/suite-runtime-flow.test.ts @@ -194,18 +194,7 @@ describe("qa suite runtime flow", () => { searchMessages: vi.fn(), waitFor: vi.fn(), }, - capabilities: { - waitForOutboundMessage: vi.fn(), - waitForCondition: vi.fn(), - getNormalizedMessageState: vi.fn(), - resetNormalizedMessageState: vi.fn(), - sendInboundMessage: vi.fn(), - injectOutboundMessage: vi.fn(), - readNormalizedMessage: vi.fn(), - executeGenericAction: vi.fn(), - waitForReady: vi.fn(), - assertNoFailureReplies: vi.fn(), - }, + waitForCondition: vi.fn(), }, repoRoot: "/repo", providerMode: "mock-openai", diff --git a/extensions/qa-lab/src/suite.ts b/extensions/qa-lab/src/suite.ts index 026ea0b73bc8..a2718f8ec3c5 100644 --- a/extensions/qa-lab/src/suite.ts +++ b/extensions/qa-lab/src/suite.ts @@ -116,22 +116,22 @@ export type QaSuiteStartLabFn = (params?: QaLabServerStartParams) => Promise Promise; outputDir: string; state: QaLabServerHandle["state"]; transportId: QaTransportId; }) { - if (params.channelDriverSelection) { - const { createQaCrablineTransportAdapter } = await import("./crabline-transport.js"); - return await createQaCrablineTransportAdapter({ + try { + return await createQaTransportAdapter({ + channelId: params.channelDriverSelection?.channel ?? params.transportId, + driver: params.channelDriverSelection ? "crabline" : params.transportId, outputDir: params.outputDir, - selection: params.channelDriverSelection, state: params.state, }); + } catch (error) { + await params.cleanupOnFailure?.().catch(() => undefined); + throw error; } - return createQaTransportAdapter({ - id: params.transportId, - state: params.state, - }); } export type QaSuiteRunParams = { @@ -700,12 +700,14 @@ async function runQaRuntimeParitySuite(params: { port: 0, embeddedGateway: "disabled", })); - const transport = await createQaSuiteTransportAdapter({ + const transportFactoryResult = await createQaSuiteTransportAdapter({ channelDriverSelection: params.channelDriverSelection, + cleanupOnFailure: ownsLab ? () => lab.stop() : undefined, outputDir: params.outputDir, state: lab.state, transportId: params.transportId, }); + const transport = transportFactoryResult.adapter; const liveScenarioOutcomes: QaLabScenarioOutcome[] = params.selectedScenarios.map((scenario) => ({ id: scenario.id, name: scenario.title, @@ -893,7 +895,7 @@ async function runQaRuntimeParitySuite(params: { watchUrl: lab.baseUrl, } satisfies QaSuiteResult; } finally { - await transport.cleanup?.(); + await transportFactoryResult.cleanup(); if (ownsLab) { await lab.stop(); } @@ -1263,12 +1265,14 @@ export async function runQaFlowSuite(params?: QaSuiteRunParams): Promise lab.stop() : undefined, outputDir, state: lab.state, transportId, }); + const transport = transportFactoryResult.adapter; const liveScenarioOutcomes: QaLabScenarioOutcome[] = selectedScenarios.map((scenario) => ({ id: scenario.id, name: scenario.title, @@ -1500,7 +1504,7 @@ export async function runQaFlowSuite(params?: QaSuiteRunParams): Promise lab.stop() : undefined, outputDir, state: lab.state, transportId, }); - writeQaSuiteProgress(progressEnabled, `provider start: ${providerMode}`); - const mock = await startQaProviderServer(providerMode); - writeQaSuiteProgress( - progressEnabled, - `provider ready: ${sanitizeQaSuiteProgressValue(mock?.baseUrl ?? "live")}`, - ); - writeQaSuiteProgress(progressEnabled, "gateway start"); - const gateway = await startQaGatewayChild({ - repoRoot, - providerBaseUrl: mock ? `${mock.baseUrl}/v1` : undefined, - transport, - transportBaseUrl: lab.listenUrl, - controlUiAllowedOrigins: [lab.listenUrl], - providerMode, - primaryModel, - alternateModel, - fastMode, - thinkingDefault: params?.thinkingDefault, - claudeCliAuthMode: params?.claudeCliAuthMode, - controlUiEnabled: params?.controlUiEnabled ?? true, - enabledPluginIds, - forwardHostHome: gatewayRuntimeOptions?.forwardHostHome, - mutateConfig: gatewayConfigPatch - ? (cfg) => applyQaMergePatch(cfg, gatewayConfigPatch) as OpenClawConfig - : undefined, - runtimeEnvPatch: mergeQaRuntimeEnvPatches( - buildQaRuntimeEnvPatch({ - providerMode, - forcedRuntime: params?.forcedRuntime, - mockBaseUrl: mock?.baseUrl, - }), - transport.createRuntimeEnvPatch?.(), - buildQaGatewayHeapCheckpointRuntimeEnvPatch(), - ), - }); - writeQaSuiteProgress( - progressEnabled, - `gateway ready: ${sanitizeQaSuiteProgressValue(gateway.baseUrl)}`, - ); - lab.setControlUi({ - controlUiProxyTarget: gateway.baseUrl, - controlUiProxyToken: gateway.token, - }); - const env: QaSuiteEnvironment = { - lab, - mock, - gateway, - // YAML scenarios should see the full staged gateway config, not just - // the transport fragment. Routing/session/plugin assertions depend on it. - cfg: gateway.cfg, - transport, - repoRoot, - providerMode, - primaryModel, - alternateModel, - webSessionIds: new Set(), - }; - + const transport = transportFactoryResult.adapter; + let mock: Awaited> | undefined; + let gateway: Awaited> | undefined; + let env: QaSuiteEnvironment | undefined; let preserveGatewayRuntimeDir: string | undefined; try { + writeQaSuiteProgress(progressEnabled, `provider start: ${providerMode}`); + const activeMock = await startQaProviderServer(providerMode); + mock = activeMock; + writeQaSuiteProgress( + progressEnabled, + `provider ready: ${sanitizeQaSuiteProgressValue(activeMock?.baseUrl ?? "live")}`, + ); + writeQaSuiteProgress(progressEnabled, "gateway start"); + const activeGateway = await startQaGatewayChild({ + repoRoot, + providerBaseUrl: activeMock ? `${activeMock.baseUrl}/v1` : undefined, + transport, + transportBaseUrl: lab.listenUrl, + controlUiAllowedOrigins: [lab.listenUrl], + providerMode, + primaryModel, + alternateModel, + fastMode, + thinkingDefault: params?.thinkingDefault, + claudeCliAuthMode: params?.claudeCliAuthMode, + controlUiEnabled: params?.controlUiEnabled ?? true, + enabledPluginIds, + forwardHostHome: gatewayRuntimeOptions?.forwardHostHome, + mutateConfig: gatewayConfigPatch + ? (cfg) => applyQaMergePatch(cfg, gatewayConfigPatch) as OpenClawConfig + : undefined, + runtimeEnvPatch: mergeQaRuntimeEnvPatches( + buildQaRuntimeEnvPatch({ + providerMode, + forcedRuntime: params?.forcedRuntime, + mockBaseUrl: activeMock?.baseUrl, + }), + transport.createRuntimeEnvPatch?.(), + buildQaGatewayHeapCheckpointRuntimeEnvPatch(), + ), + }); + gateway = activeGateway; + writeQaSuiteProgress( + progressEnabled, + `gateway ready: ${sanitizeQaSuiteProgressValue(activeGateway.baseUrl)}`, + ); + lab.setControlUi({ + controlUiProxyTarget: activeGateway.baseUrl, + controlUiProxyToken: activeGateway.token, + }); + const activeEnv: QaSuiteEnvironment = { + lab, + mock: activeMock, + gateway: activeGateway, + // YAML scenarios should see the full staged gateway config, not just + // the transport fragment. Routing/session/plugin assertions depend on it. + cfg: activeGateway.cfg, + transport, + repoRoot, + providerMode, + primaryModel, + alternateModel, + webSessionIds: new Set(), + }; + env = activeEnv; + const transportReadyTimeoutMs = resolveQaSuiteTransportReadyTimeoutMs( params?.transportReadyTimeoutMs, ); // The gateway child already waits for /readyz before returning, but the // selected transport can still be finishing account startup. Pay that // readiness cost once here so the first scenario does not race bootstrap. - await waitForTransportReady(env, transportReadyTimeoutMs).catch(async () => { - await waitForGatewayHealthy(env, transportReadyTimeoutMs); - await waitForTransportReady(env, transportReadyTimeoutMs); + await waitForTransportReady(activeEnv, transportReadyTimeoutMs).catch(async () => { + await waitForGatewayHealthy(activeEnv, transportReadyTimeoutMs); + await waitForTransportReady(activeEnv, transportReadyTimeoutMs); }); await sleep(1_000); const scenarios: QaSuiteScenarioResult[] = []; @@ -1614,7 +1626,7 @@ export async function runQaFlowSuite(params?: QaSuiteRunParams): Promise { - const gatewayProcessRssBytes = gateway.getProcessRssBytes?.() ?? null; + const gatewayProcessRssBytes = activeGateway.getProcessRssBytes?.() ?? null; if (gatewayProcessRssBytes !== null) { gatewayProcessRssSamples.push({ label, @@ -1624,7 +1636,7 @@ export async function runQaFlowSuite(params?: QaSuiteRunParams): Promise { @@ -1632,7 +1644,7 @@ export async function runQaFlowSuite(params?: QaSuiteRunParams): Promise 0 ? await captureRuntimeParityCell({ runtime: params.forcedRuntime, - gateway, + gateway: activeGateway, scenarioResult: scenarios[0], wallClockMs: Math.max(1, Date.now() - startedAt.getTime()), - mockBaseUrl: mock?.baseUrl, + mockBaseUrl: activeMock?.baseUrl, }) : undefined; const finishedAt = new Date(); @@ -1704,7 +1716,7 @@ export async function runQaFlowSuite(params?: QaSuiteRunParams): Promise