From 51f6b8f5f890aa5f79271aba7d2c71cd48044f36 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Tue, 4 Aug 2026 06:17:16 +0800 Subject: [PATCH] test(qa): cover health monitor and cache boundaries (#118961) --- .../cached-health-snapshot-boundaries.yaml | 32 ++ .../channel-health-monitor-lifecycle.yaml | 31 ++ .../gateway-rpc-account-health.yaml | 30 ++ .../cached-health-snapshot-boundaries.test.ts | 53 +++ .../cached-health-snapshot-boundaries.ts | 356 ++++++++++++++++++ ...l-health-monitor-lifecycle-runtime.test.ts | 25 ++ ...hannel-health-monitor-lifecycle-runtime.ts | 327 ++++++++++++++++ .../gateway-rpc-account-health.test.ts | 86 +++++ .../runtime/gateway-rpc-account-health.ts | 311 +++++++++++++++ 9 files changed, 1251 insertions(+) create mode 100644 qa/scenarios/observability/cached-health-snapshot-boundaries.yaml create mode 100644 qa/scenarios/observability/channel-health-monitor-lifecycle.yaml create mode 100644 qa/scenarios/observability/gateway-rpc-account-health.yaml create mode 100644 test/e2e/qa-lab/runtime/cached-health-snapshot-boundaries.test.ts create mode 100644 test/e2e/qa-lab/runtime/cached-health-snapshot-boundaries.ts create mode 100644 test/e2e/qa-lab/runtime/channel-health-monitor-lifecycle-runtime.test.ts create mode 100644 test/e2e/qa-lab/runtime/channel-health-monitor-lifecycle-runtime.ts create mode 100644 test/e2e/qa-lab/runtime/gateway-rpc-account-health.test.ts create mode 100644 test/e2e/qa-lab/runtime/gateway-rpc-account-health.ts diff --git a/qa/scenarios/observability/cached-health-snapshot-boundaries.yaml b/qa/scenarios/observability/cached-health-snapshot-boundaries.yaml new file mode 100644 index 000000000000..f64211ed79bb --- /dev/null +++ b/qa/scenarios/observability/cached-health-snapshot-boundaries.yaml @@ -0,0 +1,32 @@ +title: Cached health snapshot boundaries + +scenario: + id: cached-health-snapshot-boundaries + surface: observability + category: observability.health-and-repair + coverage: + primary: + - observability.cached-health-snapshots-kitchen-sink + - observability.cached-health-snapshots-lifecycle + - observability.cached-health-snapshots-performance + - observability.cached-health-snapshots-plugin-tools + objective: Prove health cache reuse, invalidation, live overlays, scope isolation, and plugin tool coexistence. + successCriteria: + - Fresh cache hits preserve snapshot timestamps and emit the production cached response metadata. + - Request-driven passive refresh is bounded while stale, explicit probe, and lifecycle mismatch requests refresh synchronously. + - Cached responses merge current live overlays and public refreshes omit sensitive collection. + - A real fixture plugin loads, exposes, and executes a Gateway tool without disrupting health collection. + docsRefs: + - docs/gateway/health.md + - docs/gateway/protocol.md + codeRefs: + - src/gateway/server-methods/health.ts + - src/gateway/server/health-state.ts + - test/e2e/qa-lab/runtime/cached-health-snapshot-boundaries.ts + execution: + kind: script + path: test/e2e/qa-lab/runtime/cached-health-snapshot-boundaries.ts + summary: Exercises the production health handler cache boundary and crosses a real authenticated Gateway plugin-tool invocation. + args: + - --artifact-base + - ${outputDir} diff --git a/qa/scenarios/observability/channel-health-monitor-lifecycle.yaml b/qa/scenarios/observability/channel-health-monitor-lifecycle.yaml new file mode 100644 index 000000000000..4c8f853be7ee --- /dev/null +++ b/qa/scenarios/observability/channel-health-monitor-lifecycle.yaml @@ -0,0 +1,31 @@ +title: Channel health monitor lifecycle + +scenario: + id: channel-health-monitor-lifecycle + surface: observability + category: observability.health-and-repair + coverage: + primary: + - observability.background-health-monitor-loop + - observability.startup-grace + - observability.restart-logging + objective: Prove the production channel health monitor schedules, evaluates, and repairs unhealthy account lifecycles. + successCriteria: + - No health evaluation runs before startup grace expires. + - Connect grace, stale transport, busy, and stuck snapshots follow the production health policy. + - A restart performs stop, restart-attempt reset, and start in order and emits the exact restart reason. + - Slow checks never overlap, settled checks rearm, cooldown and hourly caps bound restart attempts, and failures do not stop the loop. + - Shutdown prevents future checks and abandons restart resurrection. + docsRefs: + - docs/gateway/health.md + codeRefs: + - src/gateway/channel-health-monitor.ts + - src/gateway/channel-health-policy.ts + - test/e2e/qa-lab/runtime/channel-health-monitor-lifecycle-runtime.ts + execution: + kind: script + path: test/e2e/qa-lab/runtime/channel-health-monitor-lifecycle-runtime.ts + summary: Runs the production channel health monitor with controlled real-time account snapshots and captures lifecycle operations and logs. + args: + - --artifact-base + - ${outputDir} diff --git a/qa/scenarios/observability/gateway-rpc-account-health.yaml b/qa/scenarios/observability/gateway-rpc-account-health.yaml new file mode 100644 index 000000000000..bef68aedd252 --- /dev/null +++ b/qa/scenarios/observability/gateway-rpc-account-health.yaml @@ -0,0 +1,30 @@ +title: Gateway RPC account health + +scenario: + id: gateway-rpc-account-health + surface: observability + category: observability.health-and-repair + coverage: + primary: + - observability.gateway-rpc-health + - observability.per-account-enable-disable-settings + objective: Prove authenticated Gateway health and status RPCs reflect real per-account configuration reloads. + successCriteria: + - Authenticated health and channels.status RPCs report two running QA channel accounts, while status exposes both account summaries. + - A real CAS config.patch disables one named account and the Gateway applies the new runtime revision. + - Only the selected account configuration leaf changes. + - The default sibling remains enabled and running while the disabled account is configured, stopped, and operator-visible. + docsRefs: + - docs/gateway/health.md + - docs/gateway/protocol.md + codeRefs: + - src/gateway/server-methods/health.ts + - src/gateway/server-methods/channels.ts + - test/e2e/qa-lab/runtime/gateway-rpc-account-health.ts + execution: + kind: script + path: test/e2e/qa-lab/runtime/gateway-rpc-account-health.ts + summary: Starts a real QA bus and CLI Gateway child, performs authenticated health and status RPCs, and patches one account through config CAS and hot reload. + args: + - --artifact-base + - ${outputDir} diff --git a/test/e2e/qa-lab/runtime/cached-health-snapshot-boundaries.test.ts b/test/e2e/qa-lab/runtime/cached-health-snapshot-boundaries.test.ts new file mode 100644 index 000000000000..d8e0e33c463b --- /dev/null +++ b/test/e2e/qa-lab/runtime/cached-health-snapshot-boundaries.test.ts @@ -0,0 +1,53 @@ +import path from "node:path"; +import { describe, expect, it } from "vitest"; +import { + createFixturePlugin, + runCachedHealthSnapshotBoundariesProof, + runHandlerBoundaryProof, + withFixturePlugin, +} from "./cached-health-snapshot-boundaries.js"; + +describe("cached health snapshot boundary producer", () => { + it("proves deterministic cache reuse and invalidation boundaries", async () => { + await expect(runHandlerBoundaryProof()).resolves.toEqual({ + cacheHitSameTimestamp: true, + cachedMeta: true, + passiveRefreshBounded: true, + staleRefresh: true, + explicitProbeRefresh: true, + lifecycleMismatchRefresh: true, + liveOverlayMerged: true, + publicSensitiveOmitted: true, + }); + }); + + it("builds the fixture plugin configuration used by the real-Gateway lane", async () => { + const fixture = await createFixturePlugin(); + try { + const config = withFixturePlugin({} as never, fixture.pluginDir); + expect(config.plugins).toMatchObject({ + enabled: true, + allow: ["qa-cached-health-tool"], + entries: { "qa-cached-health-tool": { enabled: true } }, + }); + expect(config.plugins?.load?.paths).toContain(fixture.pluginDir); + } finally { + await fixture.cleanup(); + } + }); + + it.runIf(process.env.OPENCLAW_QA_REAL_GATEWAY === "1")( + "crosses the real Gateway plugin-tool boundary", + async () => { + const proof = await runCachedHealthSnapshotBoundariesProof( + path.resolve(import.meta.dirname, "../../../.."), + ); + + expect(proof.pluginLoaded).toBe(true); + expect(proof.pluginToolCataloged).toBe(true); + expect(proof.pluginToolInvoked).toBe(true); + expect(proof.healthAfterTool).toBe(true); + }, + 180_000, + ); +}); diff --git a/test/e2e/qa-lab/runtime/cached-health-snapshot-boundaries.ts b/test/e2e/qa-lab/runtime/cached-health-snapshot-boundaries.ts new file mode 100644 index 000000000000..ae5de23e25b5 --- /dev/null +++ b/test/e2e/qa-lab/runtime/cached-health-snapshot-boundaries.ts @@ -0,0 +1,356 @@ +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; +import { startQaGatewayChild } from "../../../../extensions/qa-lab/api.js"; +import type { OpenClawConfig } from "../../../../src/config/types.openclaw.js"; +import type { HealthSummary } from "../../../../src/gateway/health/types.js"; +import { healthHandlers } from "../../../../src/gateway/server-methods/health.js"; +import { createQaScriptEvidenceWriter } from "./script-evidence.js"; + +const SOURCE_PATH = "test/e2e/qa-lab/runtime/cached-health-snapshot-boundaries.ts"; +const SCENARIO_PATH = "qa/scenarios/observability/cached-health-snapshot-boundaries.yaml"; +const FIXTURE_PLUGIN_ID = "qa-cached-health-tool"; +const FIXTURE_TOOL_NAME = "qa_cached_health_echo"; +const FIXTURE_RESULT = "QA-CACHED-HEALTH-TOOL-OK"; + +type HandlerResponse = { + ok: boolean; + payload?: unknown; + error?: unknown; + meta?: Record; +}; + +type CachedHealthProof = { + cacheHitSameTimestamp: boolean; + cachedMeta: boolean; + passiveRefreshBounded: boolean; + staleRefresh: boolean; + explicitProbeRefresh: boolean; + lifecycleMismatchRefresh: boolean; + liveOverlayMerged: boolean; + publicSensitiveOmitted: boolean; + pluginLoaded: boolean; + pluginToolCataloged: boolean; + pluginToolInvoked: boolean; + healthAfterTool: boolean; +}; + +function snapshot(overrides: Partial = {}): HealthSummary { + return { + ok: true, + ts: Date.now(), + durationMs: 1, + channels: {}, + channelOrder: [], + channelLabels: {}, + heartbeatSeconds: 0, + defaultAgentId: "main", + agents: [], + sessions: { path: "state/sessions", count: 0, recent: [] }, + ...overrides, + }; +} + +export async function invokeHealthHandler(params: { + cached: HealthSummary | null; + fresh: HealthSummary; + requestParams?: Record; + runtimeSnapshot?: Record; + scopes?: string[]; + refresh?: (input: { probe: boolean; includeSensitive: boolean }) => Promise; + eventLoop?: Record; +}) { + const responses: HandlerResponse[] = []; + const refreshCalls: Array<{ probe: boolean; includeSensitive: boolean }> = []; + const refresh = params.refresh + ? params.refresh + : async (input: { probe: boolean; includeSensitive: boolean }) => { + refreshCalls.push(input); + return params.fresh; + }; + const handler = healthHandlers.health; + if (!handler) { + throw new Error("health handler is unavailable"); + } + await handler.call(healthHandlers, { + req: { type: "req", id: "qa-health", method: "health" }, + params: params.requestParams ?? {}, + respond: (ok: boolean, payload?: unknown, error?: unknown, meta?: Record) => + responses.push({ ok, payload, error, meta }), + context: { + getHealthCache: () => params.cached, + refreshHealthSnapshot: refresh, + getRuntimeSnapshot: () => params.runtimeSnapshot ?? { channels: {}, channelAccounts: {} }, + getEventLoopHealth: params.eventLoop ? () => params.eventLoop : undefined, + getConfigReloaderHotReloadStatus: () => "active", + logHealth: { error: () => undefined }, + }, + client: { connect: { role: "operator", scopes: params.scopes ?? ["operator.read"] } }, + isWebchatConnect: () => false, + } as never); + return { responses, refreshCalls }; +} + +export async function runHandlerBoundaryProof() { + const cached = snapshot({ ts: Date.now(), eventLoop: { status: "stale" } as never }); + const fresh = snapshot({ ts: cached.ts + 1 }); + const sharedRefreshCalls: Array<{ probe: boolean; includeSensitive: boolean }> = []; + const sharedRefresh = async (input: { probe: boolean; includeSensitive: boolean }) => { + sharedRefreshCalls.push(input); + return fresh; + }; + const first = await invokeHealthHandler({ + cached, + fresh, + refresh: sharedRefresh, + eventLoop: { status: "live" }, + }); + const second = await invokeHealthHandler({ + cached, + fresh, + refresh: sharedRefresh, + eventLoop: { status: "live" }, + }); + await new Promise((resolve) => setImmediate(resolve)); + + const stale = await invokeHealthHandler({ + cached: snapshot({ ts: Date.now() - 60_001 }), + fresh, + }); + const probe = await invokeHealthHandler({ + cached, + fresh, + requestParams: { probe: true }, + scopes: ["operator.read", "operator.admin"], + }); + const lifecycleCached = snapshot({ + channels: { + "qa-channel": { + accountId: "default", + running: true, + connected: true, + accounts: { + default: { accountId: "default", running: true, connected: true }, + }, + }, + }, + channelOrder: ["qa-channel"], + channelLabels: { "qa-channel": "QA Channel" }, + }); + const lifecycle = await invokeHealthHandler({ + cached: lifecycleCached, + fresh, + runtimeSnapshot: { + channels: { + "qa-channel": { accountId: "default", running: false, connected: false }, + }, + channelAccounts: { + "qa-channel": { + default: { accountId: "default", running: false, connected: false }, + }, + }, + }, + }); + const publicRefresh = await invokeHealthHandler({ + cached: null, + fresh, + scopes: ["operator.read"], + }); + + const firstResponse = first.responses[0]; + return { + cacheHitSameTimestamp: + (firstResponse?.payload as { ts?: unknown } | undefined)?.ts === cached.ts, + cachedMeta: firstResponse?.meta?.cached === true, + passiveRefreshBounded: + sharedRefreshCalls.length === 1 && second.responses[0]?.meta?.cached === true, + staleRefresh: stale.refreshCalls.length === 1 && stale.responses[0]?.meta === undefined, + explicitProbeRefresh: + probe.refreshCalls.length === 1 && + probe.refreshCalls[0]?.probe === true && + probe.refreshCalls[0]?.includeSensitive === true, + lifecycleMismatchRefresh: + lifecycle.refreshCalls.length === 1 && lifecycle.responses[0]?.meta === undefined, + liveOverlayMerged: + (firstResponse?.payload as { eventLoop?: { status?: unknown } } | undefined)?.eventLoop + ?.status === "live", + publicSensitiveOmitted: publicRefresh.refreshCalls[0]?.includeSensitive === false, + }; +} + +export async function createFixturePlugin() { + // openclaw-temp-dir: standalone producer removes this fixture root in its finally block + const root = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-cached-health-tool-")); + const pluginDir = path.join(root, FIXTURE_PLUGIN_ID); + await fs.mkdir(pluginDir, { recursive: true }); + await fs.writeFile( + path.join(pluginDir, "openclaw.plugin.json"), + `${JSON.stringify( + { + id: FIXTURE_PLUGIN_ID, + activation: { onStartup: true }, + configSchema: { type: "object", additionalProperties: false, properties: {} }, + contracts: { tools: [FIXTURE_TOOL_NAME] }, + }, + null, + 2, + )}\n`, + "utf8", + ); + await fs.writeFile( + path.join(pluginDir, "index.js"), + `module.exports = { + id: ${JSON.stringify(FIXTURE_PLUGIN_ID)}, + register(api) { + api.registerTool({ + name: ${JSON.stringify(FIXTURE_TOOL_NAME)}, + description: "Echo a bounded QA cache marker", + parameters: { + type: "object", + properties: { value: { type: "string" } }, + required: ["value"], + additionalProperties: false + }, + async execute(_toolCallId, params) { + return { content: [{ type: "text", text: ${JSON.stringify(FIXTURE_RESULT)} + ":" + params.value }] }; + } + }); + } +}; +`, + "utf8", + ); + return { pluginDir, cleanup: () => fs.rm(root, { recursive: true, force: true }) }; +} + +export function withFixturePlugin(config: OpenClawConfig, pluginDir: string): OpenClawConfig { + return { + ...config, + plugins: { + ...config.plugins, + enabled: true, + allow: [...new Set([...(config.plugins?.allow ?? []), FIXTURE_PLUGIN_ID])], + load: { + ...config.plugins?.load, + paths: [...new Set([...(config.plugins?.load?.paths ?? []), pluginDir])], + }, + entries: { + ...config.plugins?.entries, + [FIXTURE_PLUGIN_ID]: { enabled: true }, + }, + }, + }; +} + +function containsString(value: unknown, needle: string): boolean { + if (typeof value === "string") { + return value.includes(needle); + } + if (Array.isArray(value)) { + return value.some((entry) => containsString(entry, needle)); + } + if (value && typeof value === "object") { + return Object.values(value).some((entry) => containsString(entry, needle)); + } + return false; +} + +async function runPluginToolProof(repoRoot: string) { + const fixture = await createFixturePlugin(); + let gateway: Awaited> | undefined; + try { + gateway = await startQaGatewayChild({ + repoRoot, + useRepoCli: true, + transportBaseUrl: "http://127.0.0.1", + providerMode: "mock-openai", + controlUiEnabled: false, + mutateConfig: (config) => withFixturePlugin(config, fixture.pluginDir), + }); + const before = (await gateway.call("health", { probe: true })) as HealthSummary; + const catalog = await gateway.call("tools.catalog", { includePlugins: true }); + const invoked = await gateway.call("tools.invoke", { + name: FIXTURE_TOOL_NAME, + sessionKey: "agent:main:qa-cached-health", + args: { value: "after-health" }, + }); + const after = (await gateway.call("health", { probe: true })) as HealthSummary; + return { + pluginLoaded: before.plugins?.loaded.includes(FIXTURE_PLUGIN_ID) === true, + pluginToolCataloged: containsString(catalog, FIXTURE_TOOL_NAME), + pluginToolInvoked: containsString(invoked, FIXTURE_RESULT), + healthAfterTool: + after.ok === true && after.plugins?.loaded.includes(FIXTURE_PLUGIN_ID) === true, + }; + } finally { + await gateway?.stop().catch(() => undefined); + await fixture.cleanup(); + } +} + +export async function runCachedHealthSnapshotBoundariesProof( + repoRoot = path.resolve(import.meta.dirname, "../../../.."), +): Promise { + return { + ...(await runHandlerBoundaryProof()), + ...(await runPluginToolProof(repoRoot)), + }; +} + +function parseArtifactBase(argv: readonly string[]) { + const index = argv.indexOf("--artifact-base"); + const value = index >= 0 ? argv[index + 1] : undefined; + if (!value) { + throw new Error("--artifact-base is required"); + } + return path.resolve(value); +} + +export async function main(argv = process.argv.slice(2)) { + const artifactBase = parseArtifactBase(argv); + const repoRoot = path.resolve(import.meta.dirname, "../../../.."); + const writer = createQaScriptEvidenceWriter({ + artifactBase, + logFileName: "cached-health-snapshot-boundaries.log", + primaryModel: "none", + providerMode: "mock-openai", + repoRoot, + target: { + id: "cached-health-snapshot-boundaries", + title: "Cached health snapshot boundaries", + sourcePath: SCENARIO_PATH, + docsRefs: ["docs/gateway/health.md", "docs/gateway/protocol.md"], + codeRefs: [SOURCE_PATH, "src/gateway/server-methods/health.ts"], + }, + }); + const startedAt = Date.now(); + try { + const proof = await runCachedHealthSnapshotBoundariesProof(repoRoot); + const failures = Object.entries(proof) + .filter(([, passed]) => passed !== true) + .map(([name]) => `${name} failed`); + writer.appendLog(`${JSON.stringify(proof, null, 2)}\n`); + await writer.write({ + status: failures.length === 0 ? "pass" : "fail", + durationMs: Date.now() - startedAt, + details: + failures.length === 0 ? "cache and plugin-tool boundaries passed" : failures.join("; "), + }); + if (failures.length > 0) { + throw new Error(failures.join("; ")); + } + } catch (error) { + writer.appendLog(`${error instanceof Error ? error.stack : String(error)}\n`); + await writer.write({ + status: "fail", + durationMs: Date.now() - startedAt, + details: error instanceof Error ? error.message : String(error), + }); + throw error; + } +} + +if (process.argv[1] && fileURLToPath(import.meta.url) === path.resolve(process.argv[1])) { + await main(); +} diff --git a/test/e2e/qa-lab/runtime/channel-health-monitor-lifecycle-runtime.test.ts b/test/e2e/qa-lab/runtime/channel-health-monitor-lifecycle-runtime.test.ts new file mode 100644 index 000000000000..cec1064c0da6 --- /dev/null +++ b/test/e2e/qa-lab/runtime/channel-health-monitor-lifecycle-runtime.test.ts @@ -0,0 +1,25 @@ +import { describe, expect, it } from "vitest"; +import { runChannelHealthMonitorLifecycleProof } from "./channel-health-monitor-lifecycle-runtime.js"; + +describe("channel health monitor lifecycle producer", () => { + it("proves scheduler, policy, repair, logging, and shutdown boundaries", async () => { + const proof = await runChannelHealthMonitorLifecycleProof(); + + expect(proof).toMatchObject({ + graceRespected: true, + policyReasons: ["startup-connect-grace", "stale-socket", "busy", "stuck"], + singleFlight: true, + settledRearmed: true, + cooldownBounded: true, + hourlyCapLogged: true, + failureRecovered: true, + shutdownStoppedChecks: true, + }); + expect(proof.operationOrder.slice(0, 3)).toEqual([ + "stop:qa-channel:monitored:false", + "reset:qa-channel:monitored", + "start:qa-channel:monitored", + ]); + expect(proof.restartLog).toContain("restarting (reason: disconnected)"); + }); +}); diff --git a/test/e2e/qa-lab/runtime/channel-health-monitor-lifecycle-runtime.ts b/test/e2e/qa-lab/runtime/channel-health-monitor-lifecycle-runtime.ts new file mode 100644 index 000000000000..2c430ca706bf --- /dev/null +++ b/test/e2e/qa-lab/runtime/channel-health-monitor-lifecycle-runtime.ts @@ -0,0 +1,327 @@ +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; +import type { + ChannelAccountSnapshot, + ChannelId, +} from "../../../../src/channels/plugins/types.public.js"; +import { startChannelHealthMonitor } from "../../../../src/gateway/channel-health-monitor.js"; +import { + evaluateChannelHealth, + type ChannelHealthPolicy, +} from "../../../../src/gateway/channel-health-policy.js"; +import type { ChannelRuntimeSnapshot } from "../../../../src/gateway/server-channel-runtime.types.js"; +import type { ChannelManager } from "../../../../src/gateway/server-channels.js"; +import { resetLogger, setLoggerOverride } from "../../../../src/logging/logger.js"; +import { createDiagnosticLogRecordCapture } from "../../../../src/logging/test-helpers/diagnostic-log-capture.js"; +import { createQaScriptEvidenceWriter } from "./script-evidence.js"; + +const SOURCE_PATH = "test/e2e/qa-lab/runtime/channel-health-monitor-lifecycle-runtime.ts"; +const SCENARIO_PATH = "qa/scenarios/observability/channel-health-monitor-lifecycle.yaml"; + +type MonitorProof = { + graceRespected: boolean; + operationOrder: string[]; + restartLog: string; + policyReasons: string[]; + singleFlight: boolean; + settledRearmed: boolean; + cooldownBounded: boolean; + hourlyCapLogged: boolean; + failureRecovered: boolean; + shutdownStoppedChecks: boolean; +}; + +function sleep(ms: number) { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +async function waitFor(predicate: () => boolean, label: string, timeoutMs = 1_500) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (predicate()) { + return; + } + await sleep(5); + } + throw new Error(`timed out waiting for ${label}`); +} + +function snapshot(account: ChannelAccountSnapshot): ChannelRuntimeSnapshot { + return { + channels: { "qa-channel": account }, + channelAccounts: { "qa-channel": { monitored: account } }, + }; +} + +function policyReason(value: Record, policy: ChannelHealthPolicy): string { + return evaluateChannelHealth(value, policy).reason; +} + +async function captureMonitorLogs( + run: () => Promise>, +) { + // openclaw-temp-dir: standalone producer removes this log root in its finally block + const logDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-health-monitor-proof-")); + const capture = createDiagnosticLogRecordCapture(); + setLoggerOverride({ + level: "info", + consoleLevel: "silent", + file: path.join(logDir, "monitor.log"), + }); + try { + const proof = await run(); + await capture.flush(); + const messages = capture.records.map((record) => record.message); + const restartLog = + messages.find((message) => + message.includes( + "[qa-channel:monitored] health-monitor: restarting (reason: disconnected)", + ), + ) ?? ""; + return { + ...proof, + restartLog, + hourlyCapLogged: messages.some((message) => + message.includes("health-monitor: hit 1 restarts/hour limit, skipping"), + ), + }; + } finally { + capture.cleanup(); + setLoggerOverride(null); + resetLogger(); + await fs.rm(logDir, { recursive: true, force: true }); + } +} + +export async function runChannelHealthMonitorLifecycleProof(): Promise { + return await captureMonitorLogs(async () => { + const now = Date.now(); + const policy = { + channelId: "qa-channel" as ChannelId, + now, + channelConnectGraceMs: 100, + staleEventThresholdMs: 100, + }; + const policyReasons = [ + policyReason( + { running: true, connected: false, lastStartAt: now - 20, enabled: true, configured: true }, + policy, + ), + policyReason( + { + running: true, + connected: true, + lastStartAt: now - 1_000, + lastTransportActivityAt: now - 1_000, + enabled: true, + configured: true, + }, + policy, + ), + policyReason( + { + running: true, + connected: false, + busy: true, + activeRuns: 1, + lastRunActivityAt: now - 1_000, + activeRunStartedAt: now - 1_000, + enabled: true, + configured: true, + }, + policy, + ), + policyReason( + { + running: true, + connected: false, + busy: true, + activeRuns: 1, + lastRunActivityAt: now - 26 * 60_000, + activeRunStartedAt: now - 26 * 60_000, + enabled: true, + configured: true, + }, + policy, + ), + ]; + + const operations: string[] = []; + let snapshotCalls = 0; + let activeSnapshots = 0; + let maxActiveSnapshots = 0; + let failNextSnapshot = true; + let account: ChannelAccountSnapshot = { + accountId: "monitored", + running: true, + connected: false, + enabled: true, + configured: true, + lastStartAt: now - 1_000, + }; + const manager = { + getRuntimeSnapshot() { + snapshotCalls += 1; + activeSnapshots += 1; + maxActiveSnapshots = Math.max(maxActiveSnapshots, activeSnapshots); + try { + if (failNextSnapshot) { + failNextSnapshot = false; + throw new Error("controlled snapshot failure"); + } + operations.push("snapshot"); + return snapshot(account); + } finally { + activeSnapshots -= 1; + } + }, + async stopChannel(channelId: ChannelId, accountId: string, options: { manual: boolean }) { + operations.push(`stop:${channelId}:${accountId}:${String(options.manual)}`); + await sleep(35); + }, + resetRestartAttempts(channelId: ChannelId, accountId: string) { + operations.push(`reset:${channelId}:${accountId}`); + }, + async startChannel(channelId: ChannelId, accountId: string) { + operations.push(`start:${channelId}:${accountId}`); + account = { + ...account, + running: true, + connected: true, + lastStartAt: Date.now(), + lastTransportActivityAt: Date.now(), + }; + }, + getAutostartSuppression: () => null, + recoverAutostartSuppression: async () => false, + isAmbientAutostartSuppressed: () => false, + isHealthMonitorEnabled: () => true, + isManuallyStopped: () => false, + isAutoRestartScheduled: () => false, + } as unknown as ChannelManager; + + const monitor = startChannelHealthMonitor({ + channelManager: manager, + checkIntervalMs: 20, + cooldownCycles: 2, + maxRestartsPerHour: 1, + timing: { + monitorStartupGraceMs: 250, + channelConnectGraceMs: 0, + staleEventThresholdMs: 100, + }, + }); + + await sleep(40); + const graceRespected = snapshotCalls === 0; + await waitFor(() => operations.includes("start:qa-channel:monitored"), "first restart"); + await waitFor(() => snapshotCalls >= 3, "settled rearm"); + const callsAfterRecovery = snapshotCalls; + await sleep(55); + const settledRearmed = snapshotCalls > callsAfterRecovery; + const failureRecovered = operations.filter((entry) => entry === "snapshot").length >= 2; + + account = { + ...account, + connected: false, + lastStartAt: Date.now() - 1_000, + }; + await sleep(90); + const startCount = operations.filter((entry) => entry.startsWith("start:")).length; + const cooldownBounded = startCount === 1; + + monitor.shutdown(); + await monitor.waitForIdle(); + const callsAtShutdown = snapshotCalls; + await sleep(50); + + return { + graceRespected, + operationOrder: operations.filter((entry) => entry !== "snapshot"), + policyReasons, + singleFlight: maxActiveSnapshots === 1, + settledRearmed, + cooldownBounded, + failureRecovered, + shutdownStoppedChecks: snapshotCalls === callsAtShutdown, + }; + }); +} + +function parseArtifactBase(argv: readonly string[]) { + const index = argv.indexOf("--artifact-base"); + const value = index >= 0 ? argv[index + 1] : undefined; + if (!value) { + throw new Error("--artifact-base is required"); + } + return path.resolve(value); +} + +export async function main(argv = process.argv.slice(2)) { + const artifactBase = parseArtifactBase(argv); + const repoRoot = path.resolve(import.meta.dirname, "../../../.."); + const writer = createQaScriptEvidenceWriter({ + artifactBase, + logFileName: "channel-health-monitor-lifecycle.log", + primaryModel: "none", + providerMode: "mock-openai", + repoRoot, + target: { + id: "channel-health-monitor-lifecycle", + title: "Channel health monitor lifecycle", + sourcePath: SCENARIO_PATH, + docsRefs: ["docs/gateway/health.md"], + codeRefs: [SOURCE_PATH, "src/gateway/channel-health-monitor.ts"], + }, + }); + const startedAt = Date.now(); + try { + const proof = await runChannelHealthMonitorLifecycleProof(); + const expectedOrder = [ + "stop:qa-channel:monitored:false", + "reset:qa-channel:monitored", + "start:qa-channel:monitored", + ]; + const failures = [ + !proof.graceRespected && "startup grace was not respected", + proof.operationOrder.slice(0, 3).join("|") !== expectedOrder.join("|") && + `restart order was ${proof.operationOrder.join(",")}`, + !proof.restartLog && "restart reason log was not captured", + proof.policyReasons.join("|") !== + ["startup-connect-grace", "stale-socket", "busy", "stuck"].join("|") && + `policy reasons were ${proof.policyReasons.join(",")}`, + !proof.singleFlight && "checks overlapped", + !proof.settledRearmed && "settled check did not rearm", + !proof.cooldownBounded && "cooldown/hour cap allowed another restart", + !proof.hourlyCapLogged && "hourly cap warning was not captured", + !proof.failureRecovered && "loop did not recover from snapshot failure", + !proof.shutdownStoppedChecks && "checks continued after shutdown", + ].filter((value): value is string => Boolean(value)); + writer.appendLog(`${JSON.stringify(proof, null, 2)}\n`); + await writer.write({ + status: failures.length === 0 ? "pass" : "fail", + durationMs: Date.now() - startedAt, + details: + failures.length === 0 + ? "production monitor lifecycle boundaries passed" + : failures.join("; "), + }); + if (failures.length > 0) { + throw new Error(failures.join("; ")); + } + } catch (error) { + writer.appendLog(`${error instanceof Error ? error.stack : String(error)}\n`); + await writer.write({ + status: "fail", + durationMs: Date.now() - startedAt, + details: error instanceof Error ? error.message : String(error), + }); + throw error; + } +} + +if (process.argv[1] && fileURLToPath(import.meta.url) === path.resolve(process.argv[1])) { + await main(); +} diff --git a/test/e2e/qa-lab/runtime/gateway-rpc-account-health.test.ts b/test/e2e/qa-lab/runtime/gateway-rpc-account-health.test.ts new file mode 100644 index 000000000000..6be53c5826ee --- /dev/null +++ b/test/e2e/qa-lab/runtime/gateway-rpc-account-health.test.ts @@ -0,0 +1,86 @@ +import path from "node:path"; +import { describe, expect, it } from "vitest"; +import { + runGatewayRpcAccountHealthProof, + statusSummaryMentions, + withSiblingAccount, +} from "./gateway-rpc-account-health.js"; + +describe("Gateway RPC account health producer", () => { + it("adds one sibling without replacing existing QA channel config", () => { + const config = withSiblingAccount( + { + channels: { + "qa-channel": { + enabled: true, + baseUrl: "http://127.0.0.1:43124", + accounts: { existing: { enabled: false } }, + }, + }, + } as never, + "http://127.0.0.1:43125", + ); + + expect(config.channels?.["qa-channel"]).toEqual({ + enabled: true, + baseUrl: "http://127.0.0.1:43125", + botUserId: "openclaw", + botDisplayName: "OpenClaw QA", + allowFrom: ["*"], + pollTimeoutMs: 250, + accounts: { + existing: { enabled: false }, + sibling: { enabled: true }, + }, + }); + }); + + it("reads account visibility from status channel-summary lines", () => { + const status = { + channelSummary: [ + "\u001b[32mQA Channel: configured\u001b[39m", + " - default (http://127.0.0.1:43124)", + " - sibling (disabled, http://127.0.0.1:43124)", + ], + }; + + expect(statusSummaryMentions(status, "QA Channel", "default", "sibling")).toBe(true); + expect(statusSummaryMentions(status, "QA Channel", "sibling", "disabled")).toBe(true); + expect(statusSummaryMentions(status, "qa-channel")).toBe(false); + }); + + it.runIf(process.env.OPENCLAW_QA_REAL_GATEWAY === "1")( + "proves authenticated health/status RPC and a targeted account config reload", + async () => { + const proof = await runGatewayRpcAccountHealthProof( + path.resolve(import.meta.dirname, "../../../.."), + ); + + expect(proof.initialHealthOk).toBe(true); + expect(proof.initialStatusVisible).toBe(true); + expect(proof.initialAccounts.default).toMatchObject({ + enabled: true, + configured: true, + running: true, + }); + expect(proof.initialAccounts.sibling).toMatchObject({ + enabled: true, + configured: true, + running: true, + }); + expect(proof.onlyTargetConfigChanged).toBe(true); + expect(proof.finalAccounts.default).toMatchObject({ + enabled: true, + configured: true, + running: true, + }); + expect(proof.finalAccounts.sibling).toMatchObject({ + enabled: false, + configured: true, + running: false, + }); + expect(proof.finalStatusVisible).toBe(true); + }, + 180_000, + ); +}); diff --git a/test/e2e/qa-lab/runtime/gateway-rpc-account-health.ts b/test/e2e/qa-lab/runtime/gateway-rpc-account-health.ts new file mode 100644 index 000000000000..a43ee417ceca --- /dev/null +++ b/test/e2e/qa-lab/runtime/gateway-rpc-account-health.ts @@ -0,0 +1,311 @@ +import path from "node:path"; +import { fileURLToPath } from "node:url"; +import { + createQaBusState, + startQaBusServer, + startQaGatewayChild, +} from "../../../../extensions/qa-lab/api.js"; +import type { OpenClawConfig } from "../../../../src/config/types.openclaw.js"; +import { createQaScriptEvidenceWriter } from "./script-evidence.js"; + +const SOURCE_PATH = "test/e2e/qa-lab/runtime/gateway-rpc-account-health.ts"; +const SCENARIO_PATH = "qa/scenarios/observability/gateway-rpc-account-health.yaml"; +const CHANNEL_ID = "qa-channel"; +const CHANNEL_LABEL = "QA Channel"; +const TARGET_ACCOUNT_ID = "sibling"; + +type AccountState = { + accountId?: string; + configured?: boolean; + enabled?: boolean; + running?: boolean; + stateReason?: string; +}; + +type GatewayAccountHealthProof = { + initialHealthOk: boolean; + initialStatusVisible: boolean; + initialAccounts: Record; + finalAccounts: Record; + onlyTargetConfigChanged: boolean; + finalStatusVisible: boolean; +}; + +function sleep(ms: number) { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +export function withSiblingAccount(config: OpenClawConfig, baseUrl?: string): OpenClawConfig { + const channel = config.channels?.[CHANNEL_ID] as Record | undefined; + return { + ...config, + channels: { + ...config.channels, + [CHANNEL_ID]: { + ...channel, + ...(baseUrl + ? { + enabled: true, + baseUrl, + botUserId: "openclaw", + botDisplayName: "OpenClaw QA", + allowFrom: ["*"], + pollTimeoutMs: 250, + } + : {}), + accounts: { + ...((channel?.accounts as Record | undefined) ?? {}), + [TARGET_ACCOUNT_ID]: { enabled: true }, + }, + }, + }, + }; +} + +function asRecord(value: unknown): Record { + if (!value || typeof value !== "object" || Array.isArray(value)) { + throw new Error(`expected record, got ${JSON.stringify(value)}`); + } + return value as Record; +} + +function readHealthAccounts(payload: unknown): Record { + const channels = asRecord(asRecord(payload).channels); + const channel = asRecord(channels[CHANNEL_ID]); + const accounts = asRecord(channel.accounts); + return Object.fromEntries( + Object.entries(accounts).map(([accountId, state]) => [ + accountId, + asRecord(state) as AccountState, + ]), + ); +} + +function readStatusAccounts(payload: unknown): Record { + const channelAccounts = asRecord(asRecord(payload).channelAccounts); + const accounts = channelAccounts[CHANNEL_ID]; + if (!Array.isArray(accounts)) { + throw new Error(`channels.status omitted ${CHANNEL_ID} accounts`); + } + return Object.fromEntries( + accounts.map((state) => { + const record = asRecord(state) as AccountState; + return [String(record.accountId), record]; + }), + ); +} + +export function statusSummaryMentions(payload: unknown, ...needles: string[]) { + const channelSummary = asRecord(payload).channelSummary; + if (!Array.isArray(channelSummary) || channelSummary.some((line) => typeof line !== "string")) { + throw new Error(`status omitted channelSummary lines: ${JSON.stringify(payload)}`); + } + const text = channelSummary.join("\n").toLowerCase(); + return needles.every((needle) => text.includes(needle.toLowerCase())); +} + +async function waitForAccounts( + gateway: Awaited>, + predicate: (accounts: Record) => boolean, + label: string, +) { + const deadline = Date.now() + 30_000; + let latest: Record = {}; + while (Date.now() < deadline) { + const payload = await gateway.call("channels.status", { + channel: CHANNEL_ID, + probe: false, + timeoutMs: 2_000, + }); + latest = readStatusAccounts(payload); + if (predicate(latest)) { + return latest; + } + await sleep(100); + } + throw new Error(`timed out waiting for ${label}: ${JSON.stringify(latest)}`); +} + +function readChannelConfig(payload: unknown) { + const config = asRecord(asRecord(payload).config); + return structuredClone(asRecord(asRecord(config.channels)[CHANNEL_ID])); +} + +async function waitForAppliedConfig( + gateway: Awaited>, + hash: string, +) { + const deadline = Date.now() + 30_000; + while (Date.now() < deadline) { + const payload = asRecord(await gateway.call("config.get", {})); + if ( + payload.hash === hash && + typeof payload.appliedConfigHash === "string" && + payload.appliedConfigHash === payload.configRevisionHash + ) { + return payload; + } + await sleep(100); + } + throw new Error("Gateway did not apply patched account config"); +} + +export async function runGatewayRpcAccountHealthProof( + repoRoot = path.resolve(import.meta.dirname, "../../../.."), +): Promise { + const state = createQaBusState(); + const bus = await startQaBusServer({ state }); + let gateway: Awaited> | undefined; + try { + gateway = await startQaGatewayChild({ + repoRoot, + useRepoCli: true, + transportBaseUrl: bus.baseUrl, + providerMode: "mock-openai", + controlUiEnabled: false, + enabledPluginIds: [CHANNEL_ID], + mutateConfig: (config) => withSiblingAccount(config, bus.baseUrl), + }); + const initialChannelAccounts = await waitForAccounts( + gateway, + (accounts) => + accounts.default?.enabled === true && + accounts.default.running === true && + accounts[TARGET_ACCOUNT_ID]?.enabled === true && + accounts[TARGET_ACCOUNT_ID].running === true, + "both accounts running", + ); + const initialHealth = await gateway.call("health", { probe: true }); + const initialStatus = await gateway.call("status", { includeChannelSummary: true }); + const initialAccounts = readHealthAccounts(initialHealth); + const beforeConfig = asRecord(await gateway.call("config.get", {})); + const beforeChannelConfig = readChannelConfig(beforeConfig); + + const patchResult = asRecord( + await gateway.call("config.patch", { + raw: JSON.stringify({ + channels: { + [CHANNEL_ID]: { + accounts: { + [TARGET_ACCOUNT_ID]: { enabled: false }, + }, + }, + }, + }), + baseHash: beforeConfig.hash, + restartDelayMs: 0, + }), + ); + if (typeof patchResult.hash !== "string") { + throw new Error(`config.patch returned no hash: ${JSON.stringify(patchResult)}`); + } + const appliedConfig = await waitForAppliedConfig(gateway, patchResult.hash); + const afterChannelConfig = readChannelConfig(appliedConfig); + const expectedChannelConfig = structuredClone(beforeChannelConfig); + const expectedAccounts = asRecord(expectedChannelConfig.accounts); + expectedAccounts[TARGET_ACCOUNT_ID] = { + ...asRecord(expectedAccounts[TARGET_ACCOUNT_ID]), + enabled: false, + }; + + const finalChannelAccounts = await waitForAccounts( + gateway, + (accounts) => + accounts.default?.enabled === true && + accounts.default.running === true && + accounts[TARGET_ACCOUNT_ID]?.enabled === false && + accounts[TARGET_ACCOUNT_ID].running === false, + "target disabled and sibling running", + ); + const finalHealth = await gateway.call("health", { probe: true }); + const finalStatus = await gateway.call("status", { includeChannelSummary: true }); + const finalAccounts = readHealthAccounts(finalHealth); + + return { + initialHealthOk: asRecord(initialHealth).ok === true, + initialStatusVisible: + statusSummaryMentions(initialStatus, CHANNEL_LABEL, "default", TARGET_ACCOUNT_ID) && + initialChannelAccounts.default?.running === true && + initialChannelAccounts[TARGET_ACCOUNT_ID]?.running === true, + initialAccounts, + finalAccounts, + onlyTargetConfigChanged: + JSON.stringify(afterChannelConfig) === JSON.stringify(expectedChannelConfig), + finalStatusVisible: + statusSummaryMentions(finalStatus, CHANNEL_LABEL, TARGET_ACCOUNT_ID, "disabled") && + finalChannelAccounts.default?.running === true, + }; + } finally { + await gateway?.stop().catch(() => undefined); + await bus.stop().catch(() => undefined); + } +} + +function parseArtifactBase(argv: readonly string[]) { + const index = argv.indexOf("--artifact-base"); + const value = index >= 0 ? argv[index + 1] : undefined; + if (!value) { + throw new Error("--artifact-base is required"); + } + return path.resolve(value); +} + +export async function main(argv = process.argv.slice(2)) { + const artifactBase = parseArtifactBase(argv); + const repoRoot = path.resolve(import.meta.dirname, "../../../.."); + const writer = createQaScriptEvidenceWriter({ + artifactBase, + logFileName: "gateway-rpc-account-health.log", + primaryModel: "none", + providerMode: "mock-openai", + repoRoot, + target: { + id: "gateway-rpc-account-health", + title: "Gateway RPC account health", + sourcePath: SCENARIO_PATH, + docsRefs: ["docs/gateway/health.md", "docs/gateway/protocol.md"], + codeRefs: [SOURCE_PATH, "src/gateway/server-methods/health.ts"], + }, + }); + const startedAt = Date.now(); + try { + const proof = await runGatewayRpcAccountHealthProof(repoRoot); + const failures = [ + !proof.initialHealthOk && "initial health RPC was not healthy", + !proof.initialStatusVisible && "initial status did not expose both accounts", + proof.initialAccounts.default?.running !== true && + "default account was not initially running", + proof.initialAccounts[TARGET_ACCOUNT_ID]?.running !== true && + "target account was not initially running", + !proof.onlyTargetConfigChanged && "config patch changed more than the target account leaf", + proof.finalAccounts.default?.running !== true && "default sibling stopped after reload", + proof.finalAccounts[TARGET_ACCOUNT_ID]?.enabled !== false && + "target account remained enabled", + proof.finalAccounts[TARGET_ACCOUNT_ID]?.running !== false && + "target account remained running", + !proof.finalStatusVisible && "disabled target was not operator-visible", + ].filter((value): value is string => Boolean(value)); + writer.appendLog(`${JSON.stringify(proof, null, 2)}\n`); + await writer.write({ + status: failures.length === 0 ? "pass" : "fail", + durationMs: Date.now() - startedAt, + details: + failures.length === 0 ? "authenticated account health reload passed" : failures.join("; "), + }); + if (failures.length > 0) { + throw new Error(failures.join("; ")); + } + } catch (error) { + writer.appendLog(`${error instanceof Error ? error.stack : String(error)}\n`); + await writer.write({ + status: "fail", + durationMs: Date.now() - startedAt, + details: error instanceof Error ? error.message : String(error), + }); + throw error; + } +} + +if (process.argv[1] && fileURLToPath(import.meta.url) === path.resolve(process.argv[1])) { + await main(); +}