From bf1bdb429ece74655e0ebcfca1dc1273432b702b Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Tue, 4 Aug 2026 20:23:31 -0700 Subject: [PATCH] fix(qa): require ready transport accounts (#119431) --- .../src/crabline-transport-readiness.test.ts | 79 ++++++++++++++++++ extensions/qa-lab/src/crabline-transport.ts | 64 +-------------- .../qa-lab/src/qa-channel-transport.test.ts | 41 ++++++---- extensions/qa-lab/src/qa-channel-transport.ts | 75 +++-------------- extensions/qa-lab/src/qa-transport.test.ts | 50 ++++++++++++ extensions/qa-lab/src/qa-transport.ts | 81 +++++++++++++++++++ 6 files changed, 249 insertions(+), 141 deletions(-) create mode 100644 extensions/qa-lab/src/crabline-transport-readiness.test.ts diff --git a/extensions/qa-lab/src/crabline-transport-readiness.test.ts b/extensions/qa-lab/src/crabline-transport-readiness.test.ts new file mode 100644 index 000000000000..f3c455e77755 --- /dev/null +++ b/extensions/qa-lab/src/crabline-transport-readiness.test.ts @@ -0,0 +1,79 @@ +import { withTempDir } from "openclaw/plugin-sdk/test-env"; +import { describe, expect, it, vi } from "vitest"; +import { createQaBusState } from "./bus-state.js"; +import { createQaCrablineTransportAdapter } from "./crabline-transport.js"; + +async function withTelegramCrablineTransport( + run: (transport: Awaited>) => Promise, +) { + await withTempDir("qa-crabline-transport-", async (outputDir) => { + const transport = await createQaCrablineTransportAdapter({ + outputDir, + selection: { + capabilityMatrixPath: "crabline-fake-provider-capabilities.json", + channel: "telegram", + channelDriver: "crabline", + smokeArtifactPath: "crabline-fake-provider-smoke.json", + }, + state: createQaBusState(), + }); + try { + await run(transport); + } finally { + await transport.cleanup?.(); + } + }); +} + +describe("Crabline transport readiness", () => { + it("waits for the selected account to finish its channel readiness lifecycle", async () => { + await withTelegramCrablineTransport(async (transport) => { + const statuses = [ + { accountId: "other", connected: true, lifecycle: "ready", running: true }, + { + accountId: transport.accountId, + connected: false, + lifecycle: "starting", + running: true, + }, + { + accountId: transport.accountId, + connected: true, + lifecycle: "starting", + running: true, + }, + { + accountId: transport.accountId, + connected: true, + lifecycle: "blocked", + running: true, + }, + { + accountId: transport.accountId, + connected: true, + lifecycle: "ready", + restartPending: false, + running: true, + }, + ]; + const call = vi.fn(async () => ({ channelAccounts: { telegram: [statuses.shift()] } })); + + await transport.waitReady({ gateway: { call }, timeoutMs: 2_000, pollIntervalMs: 1 }); + expect(call).toHaveBeenCalledTimes(5); + }); + }); + + it("rejects a ready channel account belonging to another Crabline identity", async () => { + await withTelegramCrablineTransport(async (transport) => { + const call = vi.fn().mockResolvedValue({ + channelAccounts: { + telegram: [{ accountId: "other", connected: true, lifecycle: "ready", running: true }], + }, + }); + + await expect( + transport.waitReady({ gateway: { call }, timeoutMs: 5, pollIntervalMs: 1 }), + ).rejects.toThrow('telegram account "default" not reported; available accounts: other'); + }); + }); +}); diff --git a/extensions/qa-lab/src/crabline-transport.ts b/extensions/qa-lab/src/crabline-transport.ts index b5b7fe9ad918..733f6a2c2212 100644 --- a/extensions/qa-lab/src/crabline-transport.ts +++ b/extensions/qa-lab/src/crabline-transport.ts @@ -1,7 +1,6 @@ // Qa Lab plugin module implements Crabline local-provider transport behavior. import fs from "node:fs/promises"; import path from "node:path"; -import { setTimeout as sleep } from "node:timers/promises"; import { startOpenClawCrablineAdapter, type OpenClawCrablineChannelDriverSelection, @@ -9,7 +8,6 @@ import { type StartedOpenClawCrablineAdapter, } from "@openclaw/crabline"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; -import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; import { fetchWithSsrFGuard } from "openclaw/plugin-sdk/ssrf-runtime"; import { isRecord, @@ -23,10 +21,10 @@ import { resolveCrablineStateConversation, resolveTelegramQaSenderId, } from "./crabline-provider-targets.js"; -import { QaSuiteInfraError } from "./errors.js"; import { discardIgnoredResponseBody } from "./ignored-response-body.js"; import { QaStateBackedTransportAdapter, + waitForQaTransportAccountReady, waitForQaTransportOutboundSequence, } from "./qa-transport.js"; import type { @@ -164,64 +162,6 @@ function readTelegramLifecycleEvent(params: { }; } -async function waitForCrablineReady(params: { - accountId: string; - channel: string; - gateway: QaTransportGatewayClient; - timeoutMs?: number; - pollIntervalMs?: number; -}) { - const timeoutMs = params.timeoutMs ?? 45_000; - const pollIntervalMs = params.pollIntervalMs ?? 500; - const startedAt = Date.now(); - let lastAccountStatus = `no ${params.channel} accounts reported`; - let lastProbeError: string | null = null; - - while (Date.now() - startedAt < timeoutMs) { - try { - const payload = (await params.gateway.call( - "channels.status", - { probe: false, timeoutMs: 2_000 }, - { timeoutMs: 5_000 }, - )) as { - channelAccounts?: Record< - string, - Array<{ - accountId?: string; - running?: boolean; - restartPending?: boolean; - }> - >; - }; - const accounts = payload.channelAccounts?.[params.channel] ?? []; - const account = accounts.find((entry) => entry.accountId === params.accountId) ?? accounts[0]; - lastProbeError = null; - lastAccountStatus = account - ? JSON.stringify({ - accountId: account.accountId ?? null, - running: account.running ?? null, - restartPending: account.restartPending ?? null, - }) - : `no ${params.channel} accounts reported`; - if (account?.running && account.restartPending !== true) { - return; - } - } catch (error) { - lastProbeError = formatErrorMessage(error); - } - await sleep(pollIntervalMs); - } - - throw new QaSuiteInfraError( - "transport_ready_timeout", - [ - `timed out after ${timeoutMs}ms waiting for ${params.channel} ready`, - `last status: ${lastAccountStatus}`, - ...(lastProbeError ? [`last probe error: ${lastProbeError}`] : []), - ].join("; "), - ); -} - async function postCrablineInbound(params: { adapter: StartedOpenClawCrablineAdapter; providerInbound: OpenClawCrablineInbound; @@ -449,7 +389,7 @@ class QaCrablineTransport extends QaStateBackedTransportAdapter { timeoutMs?: number; pollIntervalMs?: number; }) => - waitForCrablineReady({ + waitForQaTransportAccountReady({ ...params, accountId: this.#adapter.accountId, channel: this.#adapter.channel, diff --git a/extensions/qa-lab/src/qa-channel-transport.test.ts b/extensions/qa-lab/src/qa-channel-transport.test.ts index 71008c79b5a2..b4e22da6e081 100644 --- a/extensions/qa-lab/src/qa-channel-transport.test.ts +++ b/extensions/qa-lab/src/qa-channel-transport.test.ts @@ -60,20 +60,31 @@ describe("qa channel transport", () => { }); }); - it("waits until the qa-channel default account is running", async () => { + it("waits until the qa-channel default account is connected and ready", async () => { const transport = createQaChannelTransport(createQaBusState()); - const call = vi - .fn() - .mockResolvedValueOnce({ - channelAccounts: { - "qa-channel": [{ accountId: "default", running: false }], - }, - }) - .mockResolvedValueOnce({ - channelAccounts: { - "qa-channel": [{ accountId: "default", running: true, restartPending: false }], - }, - }); + const statuses = [ + { accountId: "default", connected: false, lifecycle: "stopped", running: false }, + { accountId: "default", connected: false, lifecycle: "starting", running: true }, + { accountId: "default", connected: true, lifecycle: "starting", running: true }, + { accountId: "default", connected: true, lifecycle: "blocked", running: true }, + { + accountId: "default", + connected: true, + lifecycle: "ready", + restartPending: true, + running: true, + }, + { + accountId: "default", + connected: true, + lifecycle: "ready", + restartPending: false, + running: true, + }, + ]; + const call = vi.fn(async () => ({ + channelAccounts: { "qa-channel": [statuses.shift()] }, + })); await transport.waitReady({ gateway: { call }, @@ -81,7 +92,7 @@ describe("qa channel transport", () => { pollIntervalMs: 1, }); - expect(call).toHaveBeenCalledTimes(2); + expect(call).toHaveBeenCalledTimes(6); }); it("does not report another running account as the default account", async () => { @@ -112,7 +123,7 @@ describe("qa channel transport", () => { pollIntervalMs: 1, }), ).rejects.toThrow( - 'timed out after 5ms waiting for qa-channel ready; last status: {"accountId":"default","running":false,"restartPending":true}', + 'timed out after 5ms waiting for qa-channel ready; last status: {"accountId":"default","running":false,"connected":null,"lifecycle":null,"restartPending":true,"lastError":null}', ); }); diff --git a/extensions/qa-lab/src/qa-channel-transport.ts b/extensions/qa-lab/src/qa-channel-transport.ts index d8bfc542edac..b42aaba5c726 100644 --- a/extensions/qa-lab/src/qa-channel-transport.ts +++ b/extensions/qa-lab/src/qa-channel-transport.ts @@ -1,12 +1,10 @@ // Qa Lab plugin module implements qa channel transport behavior. -import { setTimeout as sleep } from "node:timers/promises"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; -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, + waitForQaTransportAccountReady, waitForQaTransportOutboundSequence, } from "./qa-transport.js"; import type { @@ -24,66 +22,6 @@ const QA_CHANNEL_ACCOUNT_ID = "default"; export const QA_CHANNEL_REQUIRED_PLUGIN_IDS = Object.freeze([QA_CHANNEL_ID]); export const QA_CHANNEL_DEFAULT_SUITE_CONCURRENCY = 4; -async function waitForQaChannelReady(params: { - gateway: QaTransportGatewayClient; - timeoutMs?: number; - pollIntervalMs?: number; -}) { - const timeoutMs = params.timeoutMs ?? 45_000; - const pollIntervalMs = params.pollIntervalMs ?? 500; - const startedAt = Date.now(); - let lastAccountStatus = "no qa-channel accounts reported"; - let lastProbeError: string | null = null; - - while (Date.now() - startedAt < timeoutMs) { - try { - const payload = (await params.gateway.call( - "channels.status", - { probe: false, timeoutMs: 2_000 }, - { timeoutMs: 5_000 }, - )) as { - channelAccounts?: Record< - string, - Array<{ - accountId?: string; - running?: boolean; - restartPending?: boolean; - }> - >; - }; - const accounts = payload.channelAccounts?.[QA_CHANNEL_ID] ?? []; - const account = accounts.find((entry) => entry.accountId === QA_CHANNEL_ACCOUNT_ID); - lastProbeError = null; - lastAccountStatus = account - ? JSON.stringify({ - accountId: account.accountId ?? null, - running: account.running ?? null, - restartPending: account.restartPending ?? null, - }) - : accounts.length > 0 - ? `qa-channel account "${QA_CHANNEL_ACCOUNT_ID}" not reported; available accounts: ${accounts - .map((entry) => entry.accountId ?? "unknown") - .join(", ")}` - : "no qa-channel accounts reported"; - if (account?.running && account.restartPending !== true) { - return; - } - } catch (error) { - lastProbeError = formatErrorMessage(error); - } - await sleep(pollIntervalMs); - } - - throw new QaSuiteInfraError( - "transport_ready_timeout", - [ - `timed out after ${timeoutMs}ms waiting for qa-channel ready`, - `last status: ${lastAccountStatus}`, - ...(lastProbeError ? [`last probe error: ${lastProbeError}`] : []), - ].join("; "), - ); -} - export function createQaChannelGatewayConfig(params: { baseUrl: string; transportPolicy?: QaTransportPolicy; @@ -171,7 +109,16 @@ class QaChannelTransport extends QaStateBackedTransportAdapter { createGatewayConfig = ({ baseUrl }: { baseUrl: string }) => createQaChannelGatewayConfig({ baseUrl, transportPolicy: this.#transportPolicy }); - waitReady = waitForQaChannelReady; + waitReady = (params: { + gateway: QaTransportGatewayClient; + timeoutMs?: number; + pollIntervalMs?: number; + }) => + waitForQaTransportAccountReady({ + ...params, + accountId: QA_CHANNEL_ACCOUNT_ID, + channel: QA_CHANNEL_ID, + }); buildAgentDelivery = ({ target }: { target: string }) => ({ channel: QA_CHANNEL_ID, replyChannel: QA_CHANNEL_ID, diff --git a/extensions/qa-lab/src/qa-transport.test.ts b/extensions/qa-lab/src/qa-transport.test.ts index 2c09f6545752..abae964d4e9a 100644 --- a/extensions/qa-lab/src/qa-transport.test.ts +++ b/extensions/qa-lab/src/qa-transport.test.ts @@ -3,9 +3,59 @@ import { describe, expect, it, vi } from "vitest"; import { createQaBusState } from "./bus-state.js"; import { createQaStateBackedTransportAdapter, + waitForQaTransportAccountReady, waitForQaTransportOutboundSequence, } from "./qa-transport.js"; +describe("waitForQaTransportAccountReady", () => { + it.each([ + { description: "disconnected", connected: false, lifecycle: "starting" }, + { description: "unauthenticated", connected: true, lifecycle: "starting" }, + { description: "blocked", connected: true, lifecycle: "blocked" }, + ])("does not declare a $description account ready", async ({ connected, lifecycle }) => { + const gateway = { + call: vi.fn().mockResolvedValue({ + channelAccounts: { + slack: [{ accountId: "sut", connected, lifecycle, running: true }], + }, + }), + }; + + await expect( + waitForQaTransportAccountReady({ + accountId: "sut", + channel: "slack", + gateway, + pollIntervalMs: 1, + timeoutMs: 5, + }), + ).rejects.toThrow(`"lifecycle":"${lifecycle}"`); + }); + + it("keeps channel-status probes inside the readiness deadline", async () => { + const call = vi.fn().mockResolvedValue({ channelAccounts: {} }); + + await expect( + waitForQaTransportAccountReady({ + accountId: "sut", + channel: "slack", + gateway: { call }, + pollIntervalMs: Number.MAX_SAFE_INTEGER, + timeoutMs: 5, + }), + ).rejects.toThrow("timed out after 5ms waiting for slack ready"); + + expect(call).toHaveBeenCalledWith( + "channels.status", + { probe: false, timeoutMs: expect.any(Number) }, + { timeoutMs: expect.any(Number) }, + ); + const [, probe, request] = call.mock.calls[0] ?? []; + expect(probe.timeoutMs).toBeLessThanOrEqual(5); + expect(request.timeoutMs).toBeLessThanOrEqual(5); + }); +}); + describe("createQaStateBackedTransportAdapter", () => { it("runs transport reset before clearing shared state", async () => { const state = createQaBusState(); diff --git a/extensions/qa-lab/src/qa-transport.ts b/extensions/qa-lab/src/qa-transport.ts index 0f99f4c77035..9e983f9e8019 100644 --- a/extensions/qa-lab/src/qa-transport.ts +++ b/extensions/qa-lab/src/qa-transport.ts @@ -1,8 +1,10 @@ // Qa Lab plugin module implements qa transport behavior. import { setTimeout as sleep } from "node:timers/promises"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; +import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; import { resolveTimerTimeoutMs } from "openclaw/plugin-sdk/number-runtime"; import type { QaRunnerCliRegistration } from "openclaw/plugin-sdk/qa-runner-runtime"; +import { QaSuiteInfraError } from "./errors.js"; import type { QaProviderMode } from "./model-selection.js"; import { extractQaFailureReplyText } from "./reply-failure.js"; import type { @@ -28,6 +30,85 @@ export type QaTransportGatewayClient = { ) => Promise; }; +export async function waitForQaTransportAccountReady(params: { + accountId: string; + channel: string; + gateway: QaTransportGatewayClient; + pollIntervalMs?: number; + timeoutMs?: number; +}): Promise { + const timeoutMs = params.timeoutMs ?? 45_000; + const pollIntervalMs = params.pollIntervalMs ?? 500; + const deadline = Date.now() + timeoutMs; + let lastAccountStatus = `no ${params.channel} accounts reported`; + let lastProbeError: string | undefined; + + while (Date.now() < deadline) { + const remainingMs = Math.max(1, deadline - Date.now()); + try { + const payload = (await params.gateway.call( + "channels.status", + { probe: false, timeoutMs: Math.min(2_000, remainingMs) }, + { timeoutMs: Math.min(5_000, remainingMs) }, + )) as { + channelAccounts?: Record< + string, + Array<{ + accountId?: string; + connected?: boolean; + lastError?: string | null; + lifecycle?: string; + restartPending?: boolean; + running?: boolean; + }> + >; + }; + const accounts = payload.channelAccounts?.[params.channel] ?? []; + const account = accounts.find((entry) => entry.accountId === params.accountId); + lastProbeError = undefined; + lastAccountStatus = account + ? JSON.stringify({ + accountId: account.accountId ?? null, + running: account.running ?? null, + connected: account.connected ?? null, + lifecycle: account.lifecycle ?? null, + restartPending: account.restartPending ?? null, + lastError: account.lastError ?? null, + }) + : accounts.length > 0 + ? `${params.channel} account "${params.accountId}" not reported; available accounts: ${accounts + .map((entry) => entry.accountId ?? "unknown") + .join(", ")}` + : `no ${params.channel} accounts reported`; + + // Connected sockets can still be unauthenticated or identity-blocked. + if ( + account?.running === true && + account.connected === true && + account.lifecycle === "ready" && + account.restartPending !== true + ) { + return; + } + } catch (error) { + lastProbeError = formatErrorMessage(error); + } + const remainingSleepMs = deadline - Date.now(); + if (remainingSleepMs > 0) { + await sleep(Math.min(pollIntervalMs, remainingSleepMs)); + } + } + + throw new QaSuiteInfraError( + "transport_ready_timeout", + [ + `timed out after ${timeoutMs}ms waiting for ${params.channel} ready`, + `last status: ${lastAccountStatus}`, + ...(lastProbeError ? [`last probe error: ${lastProbeError}`] : []), + ].join("; "), + ); +} + export type QaTransportActionName = "delete" | "edit" | "react" | "thread-create"; export type QaTransportReportParams = {