diff --git a/extensions/qa-lab/src/cli.runtime.test.ts b/extensions/qa-lab/src/cli.runtime.test.ts index 60778faca41b..0762a9d72461 100644 --- a/extensions/qa-lab/src/cli.runtime.test.ts +++ b/extensions/qa-lab/src/cli.runtime.test.ts @@ -39,7 +39,8 @@ vi.mock("./manual-lane.runtime.js", () => ({ runQaManualLane, })); -vi.mock("./suite-launch.runtime.js", () => ({ +vi.mock("./suite-launch.runtime.js", async (importOriginal) => ({ + ...(await importOriginal()), runQaFlowSuiteFromRuntime, runQaSuite, })); @@ -1481,46 +1482,44 @@ describe("qa cli runtime", () => { } }); - it("retries host suite runs once for retryable infra failures", async () => { - runQaSuite - .mockRejectedValueOnce( - new QaSuiteInfraError("agent_wait_failed", "agent.wait failed: gateway call timed out"), - ) - .mockResolvedValueOnce( - flowSuiteRuntimeResult({ - reportPath: suiteReportPath, - summaryPath: suiteSummaryPath, - }), - ); + it("leaves host suite infrastructure retries inside the suite launcher", async () => { + runQaSuite.mockRejectedValueOnce( + new QaSuiteInfraError("agent_wait_failed", "agent.wait failed: gateway call timed out"), + ); - await runQaSuiteCommand({ - repoRoot: "/tmp/openclaw-repo", - }); + await expect( + runQaSuiteCommand({ + repoRoot: "/tmp/openclaw-repo", + }), + ).rejects.toThrow("agent.wait failed: gateway call timed out"); - expect(runQaSuite).toHaveBeenCalledTimes(2); - expectWriteContains(stderrWrite, "[qa-suite] infra retry 1/1: agent.wait failed"); + expect(runQaSuite).toHaveBeenCalledTimes(1); + expect(stderrWrite.mock.calls.flat().join("")).not.toContain("[qa-suite] infra retry"); }); - it("retries host suite runs once for qa-channel readiness timeouts", async () => { - runQaSuite + it("retries host parity preflight once for qa-channel readiness timeouts", async () => { + runQaFlowSuiteFromRuntime .mockRejectedValueOnce( new QaSuiteInfraError( "transport_ready_timeout", "timed out after 180000ms waiting for qa-channel ready; last status: no qa-channel accounts reported", ), ) - .mockResolvedValueOnce( - flowSuiteRuntimeResult({ - reportPath: suiteReportPath, - summaryPath: suiteSummaryPath, - }), - ); + .mockResolvedValueOnce({ + outputDir: suiteArtifactsDir, + evidencePath: suiteEvidencePath, + watchUrl: "http://127.0.0.1:43124", + reportPath: suiteReportPath, + summaryPath: suiteSummaryPath, + scenarios: [], + }); await runQaSuiteCommand({ repoRoot: "/tmp/openclaw-repo", + preflight: true, }); - expect(runQaSuite).toHaveBeenCalledTimes(2); + expect(runQaFlowSuiteFromRuntime).toHaveBeenCalledTimes(2); expectWriteContains( stderrWrite, "[qa-suite] infra retry 1/1: timed out after 180000ms waiting for qa-channel ready", diff --git a/extensions/qa-lab/src/cli.runtime.ts b/extensions/qa-lab/src/cli.runtime.ts index 7b72c9c1d4aa..894866d97e65 100644 --- a/extensions/qa-lab/src/cli.runtime.ts +++ b/extensions/qa-lab/src/cli.runtime.ts @@ -36,7 +36,6 @@ import { } from "./coverage-report.js"; import { buildQaDockerHarnessImage, writeQaDockerHarnessFiles } from "./docker-harness.js"; import { runQaDockerUp } from "./docker-up.runtime.js"; -import { QaSuiteArtifactError, QaSuiteInfraError } from "./errors.js"; import type { QaCliBackendAuthMode } from "./gateway-child.js"; import { createMockJsonlReplayCellRunner, @@ -94,7 +93,11 @@ import { type QaScorecardEvidenceMode, } from "./scorecard-taxonomy.js"; import { isQaSelfCheckSuccessful } from "./self-check.js"; -import { runQaFlowSuiteFromRuntime, runQaSuite } from "./suite-launch.runtime.js"; +import { + runQaFlowSuiteFromRuntime, + runQaSuite, + runQaSuiteWithInfraRetry, +} from "./suite-launch.runtime.js"; import { resolveQaSuiteScenarioChannel, resolveQaSuiteScenarioChannels } from "./suite-planning.js"; import { readQaSuiteFailedOrSkippedScenarioCountFromFile } from "./suite-summary.js"; import { @@ -108,16 +111,8 @@ import { type QaToolCoverageSuiteSummary, } from "./tool-coverage-report.js"; -const QA_SUITE_INFRA_RETRY_LIMIT = 1; const QA_CREDENTIAL_PAYLOAD_MAX_BYTES_ENV = "OPENCLAW_QA_CREDENTIAL_PAYLOAD_MAX_BYTES"; const DEFAULT_QA_CREDENTIAL_PAYLOAD_MAX_BYTES = 64 * 1024 * 1024; -const QA_SUITE_INFRA_RETRY_NETWORK_ERROR_CODES = new Set([ - "ECONNRESET", - "ECONNREFUSED", - "EPIPE", - "ETIMEDOUT", - "UND_ERR_SOCKET", -]); type InterruptibleServer = { baseUrl: string; stop(): Promise; @@ -345,51 +340,6 @@ function rejectNonFlowScenarioIds(params: { } } -function isQaSuiteInfraRetryableError(error: unknown) { - if (error instanceof QaSuiteArtifactError || error instanceof QaSuiteInfraError) { - return true; - } - return hasQaSuiteRetryableNetworkCode(error); -} - -function hasQaSuiteRetryableNetworkCode(error: unknown) { - let current: unknown = error; - for (let depth = 0; depth < 4 && current; depth += 1) { - if (typeof current !== "object") { - return false; - } - const record = current as { cause?: unknown; code?: unknown }; - if ( - typeof record.code === "string" && - QA_SUITE_INFRA_RETRY_NETWORK_ERROR_CODES.has(record.code.toUpperCase()) - ) { - return true; - } - current = record.cause; - } - return false; -} - -async function runQaSuiteWithInfraRetry( - run: () => Promise, - maxRetries = QA_SUITE_INFRA_RETRY_LIMIT, -) { - for (let attempt = 0; attempt <= maxRetries; attempt += 1) { - try { - return await run(); - } catch (error) { - const retryable = isQaSuiteInfraRetryableError(error); - if (!retryable || attempt >= maxRetries) { - throw error; - } - process.stderr.write( - `[qa-suite] infra retry ${attempt + 1}/${maxRetries}: ${formatErrorMessage(error)}\n`, - ); - } - } - throw new Error("unreachable qa suite retry state"); -} - async function runQaParityPreflight(params: { repoRoot: string; transportId: QaTransportId; @@ -1023,45 +973,43 @@ export async function runQaSuiteCommand(opts: QaSuiteCommandOptions) { return undefined; } const thinkingDefault = parseQaThinkingLevel("--thinking", opts.thinking); - const runtimeResult = await runQaSuiteWithInfraRetry(() => - runQaSuite({ - repoRoot, - outputDir: resolveRepoRelativeOutputDir(repoRoot, opts.outputDir), - evidenceMode: opts.evidenceMode, - transportId, - channelDriver, - ...(liveAdapterFactories - ? { - adapterFactories: liveAdapterFactories, - ...(liveChannelId ? { channelId: liveChannelId } : {}), - adapterOptions: { - repoRoot, - sutAccountId: opts.sutAccountId, - credentialSource: opts.credentialSource, - credentialRole: opts.credentialRole, - explicitScenarioSelection: - opts.explicitScenarioSelection ?? Boolean(opts.scenarioIds?.length), - }, - } + const runtimeResult = await runQaSuite({ + repoRoot, + outputDir: resolveRepoRelativeOutputDir(repoRoot, opts.outputDir), + evidenceMode: opts.evidenceMode, + transportId, + channelDriver, + ...(liveAdapterFactories + ? { + adapterFactories: liveAdapterFactories, + ...(liveChannelId ? { channelId: liveChannelId } : {}), + adapterOptions: { + repoRoot, + sutAccountId: opts.sutAccountId, + credentialSource: opts.credentialSource, + credentialRole: opts.credentialRole, + explicitScenarioSelection: + opts.explicitScenarioSelection ?? Boolean(opts.scenarioIds?.length), + }, + } + : {}), + channelDriverSelection, + ...(opts.providerMode !== undefined ? { providerMode } : {}), + primaryModel, + alternateModel, + fastMode: opts.fastMode, + failFast: opts.failFast, + ...(thinkingDefault ? { thinkingDefault } : {}), + ...(claudeCliAuthMode ? { claudeCliAuthMode } : {}), + scenarioIds: liveChannelId ? scenarioIds : hostScenarioIds, + ...(opts.enabledPluginIds !== undefined ? { enabledPluginIds: opts.enabledPluginIds } : {}), + ...(liveChannelId + ? { concurrency: 1 } + : opts.concurrency !== undefined + ? { concurrency: parseQaPositiveIntegerOption("--concurrency", opts.concurrency) } : {}), - channelDriverSelection, - ...(opts.providerMode !== undefined ? { providerMode } : {}), - primaryModel, - alternateModel, - fastMode: opts.fastMode, - failFast: opts.failFast, - ...(thinkingDefault ? { thinkingDefault } : {}), - ...(claudeCliAuthMode ? { claudeCliAuthMode } : {}), - scenarioIds: liveChannelId ? scenarioIds : hostScenarioIds, - ...(opts.enabledPluginIds !== undefined ? { enabledPluginIds: opts.enabledPluginIds } : {}), - ...(liveChannelId - ? { concurrency: 1 } - : opts.concurrency !== undefined - ? { concurrency: parseQaPositiveIntegerOption("--concurrency", opts.concurrency) } - : {}), - ...(runtimePair ? { runtimePair } : {}), - }), - ); + ...(runtimePair ? { runtimePair } : {}), + }); switch (runtimeResult.executionKind) { case "suite": { const result = runtimeResult.result; diff --git a/extensions/qa-lab/src/suite-launch.runtime.test.ts b/extensions/qa-lab/src/suite-launch.runtime.test.ts index b46e7c3584cd..3c30e2d51df4 100644 --- a/extensions/qa-lab/src/suite-launch.runtime.test.ts +++ b/extensions/qa-lab/src/suite-launch.runtime.test.ts @@ -2,6 +2,7 @@ import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { QaSuiteInfraError } from "./errors.js"; import type { QaLabServerHandle } from "./lab-server.types.js"; import type { QaSuiteScenarioResult } from "./suite.js"; import type { @@ -77,6 +78,28 @@ function trackMaxActiveFlowRuns() { return () => maxActive; } +function mockFlowPartitionFailures(failuresByScenarioId: ReadonlyMap) { + const run = runQaFlowSuite.getMockImplementation(); + if (!run) { + throw new Error("expected default QA flow suite mock implementation"); + } + const attempts = new Map(); + runQaFlowSuite.mockImplementation(async (params) => { + const scenarioId = params?.scenarioIds?.[0]; + if (!scenarioId) { + throw new Error("expected one scenario per flow partition"); + } + const attempt = (attempts.get(scenarioId) ?? 0) + 1; + attempts.set(scenarioId, attempt); + const failure = failuresByScenarioId.get(scenarioId)?.[attempt - 1]; + if (failure) { + throw failure; + } + return await run(params); + }); + return attempts; +} + describe("qa suite runtime launcher", () => { beforeEach(() => { runQaFlowSuite.mockReset(); @@ -180,6 +203,34 @@ describe("qa suite runtime launcher", () => { expect(runQaTestFileScenarios).not.toHaveBeenCalled(); }); + it("retries a flow-only suite once for retryable infrastructure failures", async () => { + const attempts = mockFlowPartitionFailures( + new Map([ + [ + "channel-chat-baseline", + [new QaSuiteInfraError("agent_wait_failed", "agent.wait failed")], + ], + ]), + ); + const stderrWrite = vi.spyOn(process.stderr, "write").mockReturnValue(true); + + try { + const result = await runQaSuite({ + repoRoot: process.cwd(), + providerMode: "mock-openai", + scenarioIds: ["channel-chat-baseline"], + }); + + expect(result.executionKind).toBe("flow"); + expect(attempts.get("channel-chat-baseline")).toBe(2); + expect(stderrWrite.mock.calls.flat().join("")).toContain( + "[qa-suite] infra retry 1/1: agent.wait failed", + ); + } finally { + stderrWrite.mockRestore(); + } + }); + it("partitions flow-only suites that request isolated workers", async () => { const repoRoot = await makeTempRepo("qa-suite-flow-only-isolated-"); const result = await runQaSuite({ @@ -290,6 +341,107 @@ describe("qa suite runtime launcher", () => { ); }); + it("retries only the failed channel partition in a mixed-channel suite", async () => { + const repoRoot = await makeTempRepo("qa-suite-partition-retry-"); + const attempts = mockFlowPartitionFailures( + new Map([ + [ + "whatsapp-status-command", + [new QaSuiteInfraError("transport_ready_timeout", "WhatsApp readiness timed out")], + ], + ]), + ); + + const result = await runQaSuite({ + repoRoot, + outputDir: ".artifacts/qa-e2e/partition-retry", + providerMode: "mock-openai", + channelDriver: "live", + adapterFactories: [{ id: "portable-driver", matches: () => true, create: vi.fn() }], + concurrency: 4, + scenarioIds: [ + "telegram-help-command", + "matrix-restart-resume", + "slack-canary", + "whatsapp-status-command", + ], + }); + + expect(result.executionKind).toBe("suite"); + expect(Object.fromEntries(attempts)).toEqual({ + "telegram-help-command": 1, + "matrix-restart-resume": 1, + "slack-canary": 1, + "whatsapp-status-command": 2, + }); + expect(result.result.scenarios).toHaveLength(4); + expect(new Set(result.result.scenarios.map((scenario) => scenario.name))).toEqual( + new Set([ + "telegram-help-command", + "matrix-restart-resume", + "slack-canary", + "whatsapp-status-command", + ]), + ); + }); + + it("does not retry mixed-channel partitions for generic timeout wording", async () => { + const repoRoot = await makeTempRepo("qa-suite-partition-generic-timeout-"); + const attempts = mockFlowPartitionFailures( + new Map([ + [ + "whatsapp-status-command", + [new Error("approval-turn timed out waiting for post-approval read")], + ], + ]), + ); + + await expect( + runQaSuite({ + repoRoot, + outputDir: ".artifacts/qa-e2e/partition-generic-timeout", + providerMode: "mock-openai", + channelDriver: "live", + adapterFactories: [{ id: "portable-driver", matches: () => true, create: vi.fn() }], + concurrency: 2, + scenarioIds: ["telegram-help-command", "whatsapp-status-command"], + }), + ).rejects.toThrow("approval-turn timed out waiting for post-approval read"); + + expect(attempts.get("telegram-help-command")).toBe(1); + expect(attempts.get("whatsapp-status-command")).toBe(1); + }); + + it("preserves completed partitions when a retryable channel fails twice", async () => { + const repoRoot = await makeTempRepo("qa-suite-partition-retry-exhausted-"); + const attempts = mockFlowPartitionFailures( + new Map([ + [ + "whatsapp-status-command", + [ + new QaSuiteInfraError("transport_ready_timeout", "WhatsApp readiness timed out"), + new QaSuiteInfraError("transport_ready_timeout", "WhatsApp readiness timed out again"), + ], + ], + ]), + ); + + await expect( + runQaSuite({ + repoRoot, + outputDir: ".artifacts/qa-e2e/partition-retry-exhausted", + providerMode: "mock-openai", + channelDriver: "live", + adapterFactories: [{ id: "portable-driver", matches: () => true, create: vi.fn() }], + concurrency: 2, + scenarioIds: ["telegram-help-command", "whatsapp-status-command"], + }), + ).rejects.toThrow("WhatsApp readiness timed out again"); + + expect(attempts.get("telegram-help-command")).toBe(1); + expect(attempts.get("whatsapp-status-command")).toBe(2); + }); + it("runs distinct pluggable-driver channels within the global concurrency budget", async () => { const repoRoot = await makeTempRepo("qa-suite-pluggable-channel-concurrency-"); const maxActive = trackMaxActiveFlowRuns(); diff --git a/extensions/qa-lab/src/suite-launch.runtime.ts b/extensions/qa-lab/src/suite-launch.runtime.ts index bdc43ab0e77c..0693109978fb 100644 --- a/extensions/qa-lab/src/suite-launch.runtime.ts +++ b/extensions/qa-lab/src/suite-launch.runtime.ts @@ -3,6 +3,7 @@ import fs from "node:fs/promises"; import path from "node:path"; import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; import { isRepoRootRelativeRef, toRepoRelativePath } from "./cli-paths.js"; +import { QaSuiteArtifactError, QaSuiteInfraError } from "./errors.js"; import { QA_EVIDENCE_FILENAME, QA_EVIDENCE_SUMMARY_KIND, @@ -81,6 +82,14 @@ type QaSuiteExecutionPlan = const MAX_SHARED_FLOW_PARTITIONS = 4; const MAX_ISOLATED_FLOW_CONCURRENCY = 8; const ISOLATED_FLOW_WORKER_START_STAGGER_MS = 1_500; +const QA_SUITE_INFRA_RETRY_LIMIT = 1; +const QA_SUITE_INFRA_RETRY_NETWORK_ERROR_CODES = new Set([ + "ECONNRESET", + "ECONNREFUSED", + "EPIPE", + "ETIMEDOUT", + "UND_ERR_SOCKET", +]); const CREDENTIAL_POOL_UNAVAILABLE_CODES = new Set(["NO_CREDENTIAL_AVAILABLE", "POOL_EXHAUSTED"]); type QaUnifiedPartitionResult = { @@ -107,6 +116,50 @@ type QaFlowChannelGroup = { scenarios: QaSeedScenarioWithSource[]; }; +function hasQaSuiteRetryableNetworkCode(error: unknown) { + let current: unknown = error; + for (let depth = 0; depth < 4 && current; depth += 1) { + if (typeof current !== "object") { + return false; + } + const record = current as { cause?: unknown; code?: unknown }; + if ( + typeof record.code === "string" && + QA_SUITE_INFRA_RETRY_NETWORK_ERROR_CODES.has(record.code.toUpperCase()) + ) { + return true; + } + current = record.cause; + } + return false; +} + +function isQaSuiteInfraRetryableError(error: unknown) { + if (error instanceof QaSuiteArtifactError || error instanceof QaSuiteInfraError) { + return true; + } + return hasQaSuiteRetryableNetworkCode(error); +} + +export async function runQaSuiteWithInfraRetry( + run: () => Promise, + maxRetries = QA_SUITE_INFRA_RETRY_LIMIT, +) { + for (let attempt = 0; attempt <= maxRetries; attempt += 1) { + try { + return await run(); + } catch (error) { + if (!isQaSuiteInfraRetryableError(error) || attempt >= maxRetries) { + throw error; + } + process.stderr.write( + `[qa-suite] infra retry ${attempt + 1}/${maxRetries}: ${formatErrorMessage(error)}\n`, + ); + } + } + throw new Error("unreachable qa suite retry state"); +} + async function loadQaLabServerRuntime() { const { startQaLabServer } = await import("./lab-server.js"); return startQaLabServer; @@ -1022,12 +1075,19 @@ async function runUnifiedQaSuite(params: { ); return partition.startedScenarioIds.some((scenarioId) => !returnedScenarioIds.has(scenarioId)); }; - const runPartitionTasks = async (tasks: readonly QaUnifiedPartitionTask[], maxWeight: number) => - failFast - ? await mapQaSuiteWithConcurrency(tasks, 1, runFailFastPartition, { + const runPartitionTasks = async (tasks: readonly QaUnifiedPartitionTask[], maxWeight: number) => { + // Retry inside the scheduled task so its weight and exclusive key stay held; + // one failed channel must not replay partitions that already completed. + const retryingTasks = tasks.map((task) => ({ + ...task, + run: async () => await runQaSuiteWithInfraRetry(task.run), + })); + return failFast + ? await mapQaSuiteWithConcurrency(retryingTasks, 1, runFailFastPartition, { shouldStop: partitionFailed, }) - : await runWeightedUnifiedPartitionTasks(tasks, maxWeight); + : await runWeightedUnifiedPartitionTasks(retryingTasks, maxWeight); + }; const concurrentPartitionResults = await runPartitionTasks(concurrentPartitionTasks, concurrency); // Script scenarios may rebuild the checkout's shared dist tree. Wait until every // flow Gateway has stopped so package postbuild cannot invalidate its loaded chunks. @@ -1129,7 +1189,7 @@ export async function runQaSuite(...args: [QaSuiteRunParams?]): Promise runQaFlowSuiteFromRuntime(...args)), }; }