fix(qa): require ready transport accounts (#119431)

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<ReturnType<typeof createQaCrablineTransportAdapter>>) => Promise<void>,
+) {
+  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');
+    });
+  });
+});

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,

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}',
     );
   });
 

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,

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();

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<unknown>;
 };
 
+export async function waitForQaTransportAccountReady(params: {
+  accountId: string;
+  channel: string;
+  gateway: QaTransportGatewayClient;
+  pollIntervalMs?: number;
+  timeoutMs?: number;
+}): Promise<void> {
+  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 = {