ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-24 03:15:46 -06:00
Code Issues Packages Projects Releases Wiki Activity

feat(gateway): track event-loop degradation persistence and add concurrency benchmark (#118193)

Browse Source 
* feat(gateway): persist event loop degradation metrics

* fix(gateway): bound concurrency benchmark turn waits
This commit is contained in:
Peter Steinberger
2026-08-02 14:21:44 -07:00
committed by GitHub
parent c79dcff608
commit 3f3ceb2def
30 changed files with 1275 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
package.json
+1
View File
@@ -1836,6 +1836,7 @@
"test:force": "node --import tsx scripts/test-force.ts",
"test:gateway": "node scripts/run-with-env.mjs OPENCLAW_GATEWAY_PROJECT_SHARDS=1 -- node scripts/run-vitest.mjs run --config test/vitest/vitest.gateway.config.ts",
"test:gateway:cpu-scenarios": "node scripts/check-gateway-cpu-scenarios.mjs",
"test:gateway:concurrency": "node --import tsx scripts/bench-gateway-concurrency.ts",
"test:gateway:memory-fd-repro": "node scripts/check-memory-fd-repro.mjs",
"test:gateway:watch-regression": "node scripts/check-gateway-watch-regression.mjs",
"test:install:e2e": "bash scripts/test-install-sh-e2e-docker.sh",
packages/gateway-protocol/src/channels.schema.test.ts
+1
View File
@@ -71,6 +71,7 @@ describe("ChannelsStatusResultSchema", () => {
warnings: ["discord:default probe timed out after 1000ms"],
eventLoop: {
degraded: true,
degradedSinceMs: 61_000,
reasons: ["event_loop_delay", "cpu"],
intervalMs: 62_000,
delayP99Ms: 1_250.5,
packages/gateway-protocol/src/schema/channels.ts
+1
View File
@@ -701,6 +701,7 @@ const ChannelUiMetaSchema = closedObject({
/** Event-loop health snapshot included with channel status responses. */
const ChannelEventLoopHealthSchema = closedObject({
degraded: Type.Boolean(),
degradedSinceMs: Type.Optional(Type.Union([Type.Integer({ minimum: 0 }), Type.Null()])),
reasons: Type.Array(
Type.Union([
Type.Literal("event_loop_delay"),
packages/gateway-protocol/src/schema/snapshot.test.ts
+20
View File
@@ -40,4 +40,24 @@ describe("SnapshotSchema", () => {
),
).toBe(true);
});
it("accepts persistent event-loop health duration", () => {
const snapshot = {
...snapshotWithPresence({ ts: 1 }),
health: {
eventLoop: {
degraded: true,
degradedSinceMs: 61_000,
reasons: ["event_loop_delay"],
intervalMs: 30_000,
delayP99Ms: 1_200,
delayMaxMs: 1_500,
utilization: 0.75,
cpuCoreRatio: 0.5,
},
},
};
expect(Value.Check(SnapshotSchema, snapshot)).toBe(true);
});
});
packages/gateway-protocol/src/schema/snapshot.ts
+1
View File
@@ -62,6 +62,7 @@ const HealthSnapshotSchema = closedObject({
eventLoop: Type.Optional(
closedObject({
degraded: Type.Boolean(),
degradedSinceMs: Type.Optional(Type.Union([Type.Integer({ minimum: 0 }), Type.Null()])),
reasons: Type.Array(
Type.Union([
Type.Literal("event_loop_delay"),
packages/terminal-core/src/health-style.ts
+3
View File
@@ -25,6 +25,9 @@ export function styleHealthChannelLine(line: string, rich: boolean): string {
if (normalized.startsWith("failed")) {
return applyPrefix("failed", theme.error);
}
if (normalized.startsWith("degraded")) {
return applyPrefix("degraded", theme.warn);
}
if (normalized.startsWith("ok")) {
return applyPrefix("ok", theme.success);
}
scripts/bench-gateway-concurrency.ts
+719
View File
@@ -0,0 +1,719 @@
import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process";
// Bench Gateway Concurrency script measures gateway probes during synthetic streaming turns.
import { randomUUID } from "node:crypto";
import { copyFileSync, mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
import { request } from "node:http";
import { tmpdir } from "node:os";
import path from "node:path";
import { performance } from "node:perf_hooks";
import { pathToFileURL } from "node:url";
import { PROTOCOL_VERSION } from "../packages/gateway-protocol/src/index.js";
import { applyMockOpenAiModelConfig } from "./e2e/lib/fixtures/mock-openai-config.mjs";
import { delay, stopChild } from "./lib/gateway-bench-child.ts";
import { getFreePort } from "./lib/gateway-bench-probes.ts";
import {
BASE_GATEWAY_BENCH_CONFIG,
buildGatewayBenchChildArgs,
CliArgumentError,
createGatewayBenchEnv,
hasFlag,
hasHelpFlag,
parseFlagValue,
parseNonNegativeInt,
parsePositiveInt,
resolveEntry,
resolveOutputPath,
validateCliArgs,
waitForInitialProbe,
writeGatewayBenchConfig,
} from "./lib/gateway-bench-runtime.ts";
import { createGatewayWsClient } from "./lib/gateway-ws-client.ts";
type MetricSummary = {
count: number;
max: number;
p50: number;
p95: number;
p99: number;
};
type TimedProbe = {
atMs: number;
latencyMs: number;
ok: boolean;
};
type ReadyProbe = TimedProbe & {
cpuCoreRatio: number | null;
degraded: boolean | null;
degradedSinceMs: number | null;
delayP99Ms: number | null;
status: number;
utilization: number | null;
};
type GatewayRpc = <T>(method: string, params: unknown, timeoutMs?: number) => Promise<T>;
type BenchmarkRun = {
controlUi: TimedProbe[];
durationMs: number;
readyz: ReadyProbe[];
sessionsList: TimedProbe[];
turnCount: number;
turnsDurationMs: number;
};
type CliOptions = {
cadenceMs: number;
concurrency: number;
entry: string;
json: boolean;
output?: string;
runs: number;
timeoutMs: number;
warmup: number;
};
const DEFAULT_CADENCE_MS = 100;
const DEFAULT_CONCURRENCY = 8;
const DEFAULT_ENTRY = "dist/entry.js";
const DEFAULT_RUNS = 1;
const DEFAULT_TIMEOUT_MS = 120_000;
const DEFAULT_WARMUP = 0;
const MOCK_RESPONSE_CHUNK_DELAY_MS = 1_000;
const MAX_CONCURRENCY = 64;
const MAX_RUNS = 20;
const MAX_WARMUP = 10;
const MAX_SAMPLES_PER_RUN = 2_048;
const MAX_HTTP_BODY_BYTES = 1_048_576;
const HTTP_TIMEOUT_MS = 20_000;
const AGENT_WAIT_RPC_GRACE_MS = 5_000;
const BOOLEAN_FLAGS = new Set(["--help", "-h", "--json"]);
const VALUE_FLAGS = new Set([
"--cadence-ms",
"--concurrency",
"--entry",
"--output",
"--runs",
"--timeout-ms",
"--warmup",
]);
function parseBoundedPositiveInt(
raw: string | undefined,
fallback: number,
label: string,
max: number,
): number {
const value = parsePositiveInt(raw, fallback, label);
if (value > max) {
throw new CliArgumentError(`${label} must be at most ${max}`);
}
return value;
}
function parseBoundedNonNegativeInt(
raw: string | undefined,
fallback: number,
label: string,
max: number,
): number {
const value = parseNonNegativeInt(raw, fallback, label);
if (value > max) {
throw new CliArgumentError(`${label} must be at most ${max}`);
}
return value;
}
function parseOptions(argv: string[] = process.argv.slice(2)): CliOptions {
validateCliArgs(argv, { booleanFlags: BOOLEAN_FLAGS, valueFlags: VALUE_FLAGS });
return {
cadenceMs: parseBoundedPositiveInt(
parseFlagValue(argv, "--cadence-ms"),
DEFAULT_CADENCE_MS,
"--cadence-ms",
5_000,
),
concurrency: parseBoundedPositiveInt(
parseFlagValue(argv, "--concurrency"),
DEFAULT_CONCURRENCY,
"--concurrency",
MAX_CONCURRENCY,
),
entry: resolveEntry(parseFlagValue(argv, "--entry"), DEFAULT_ENTRY),
json: hasFlag(argv, "--json"),
output: resolveOutputPath(parseFlagValue(argv, "--output")),
runs: parseBoundedPositiveInt(parseFlagValue(argv, "--runs"), DEFAULT_RUNS, "--runs", MAX_RUNS),
timeoutMs: parseBoundedPositiveInt(
parseFlagValue(argv, "--timeout-ms"),
DEFAULT_TIMEOUT_MS,
"--timeout-ms",
10 * 60_000,
),
warmup: parseBoundedNonNegativeInt(
parseFlagValue(argv, "--warmup"),
DEFAULT_WARMUP,
"--warmup",
MAX_WARMUP,
),
};
}
function printUsage(): void {
console.log(`OpenClaw Gateway concurrency benchmark
Usage:
pnpm test:gateway:concurrency -- [options]
node --import tsx scripts/bench-gateway-concurrency.ts [options]
Options:
--concurrency <n> Concurrent synthetic streaming turns (default: ${DEFAULT_CONCURRENCY})
--runs <n> Measured gateway runs (default: ${DEFAULT_RUNS})
--warmup <n> Warmup gateway runs (default: ${DEFAULT_WARMUP})
--cadence-ms <ms> Probe cadence (default: ${DEFAULT_CADENCE_MS})
--timeout-ms <ms> Whole benchmark wall-clock cap (default: ${DEFAULT_TIMEOUT_MS})
--entry <path> Gateway CLI entry file (default: ${DEFAULT_ENTRY})
--output <path> Write machine-readable JSON to a file
--json Emit machine-readable JSON
--help, -h Show this text
`);
}
function percentile(sorted: readonly number[], percentileValue: number): number {
const index = Math.max(
0,
Math.min(sorted.length - 1, Math.ceil((percentileValue / 100) * sorted.length) - 1),
);
return sorted[index] ?? 0;
}
function summarizeNumbers(values: readonly number[]): MetricSummary | null {
const sorted = values.filter(Number.isFinite).toSorted((a, b) => a - b);
if (sorted.length === 0) {
return null;
}
return {
count: sorted.length,
max: sorted.at(-1) ?? 0,
p50: percentile(sorted, 50),
p95: percentile(sorted, 95),
p99: percentile(sorted, 99),
};
}
function remainingMs(deadlineAt: number): number {
return Math.max(0, deadlineAt - performance.now());
}
function requireRemainingMs(deadlineAt: number, label: string): number {
const remaining = remainingMs(deadlineAt);
if (remaining <= 0) {
throw new Error(`benchmark timed out while ${label}`);
}
return remaining;
}
async function requestHttp(params: {
accept: string;
deadlineAt: number;
path: string;
port: number;
}): Promise<{ body: string; latencyMs: number; status: number }> {
const startedAt = performance.now();
const timeoutMs = Math.max(
1,
Math.min(HTTP_TIMEOUT_MS, requireRemainingMs(params.deadlineAt, `requesting ${params.path}`)),
);
return await new Promise((resolve, reject) => {
const req = request(
{
headers: { accept: params.accept },
host: "127.0.0.1",
method: "GET",
path: params.path,
port: params.port,
timeout: timeoutMs,
},
(res) => {
const chunks: Buffer[] = [];
let bytes = 0;
res.on("data", (chunk: Buffer) => {
bytes += chunk.length;
if (bytes > MAX_HTTP_BODY_BYTES) {
req.destroy(new Error(`${params.path} response exceeded ${MAX_HTTP_BODY_BYTES} bytes`));
return;
}
chunks.push(chunk);
});
res.on("end", () => {
resolve({
body: Buffer.concat(chunks).toString("utf8"),
latencyMs: performance.now() - startedAt,
status: res.statusCode ?? 0,
});
});
},
);
req.once("error", reject);
req.once("timeout", () => req.destroy(new Error(`${params.path} request timed out`)));
req.end();
});
}
function numberOrNull(value: unknown): number | null {
return typeof value === "number" && Number.isFinite(value) ? value : null;
}
function captureChildOutput(child: ChildProcessWithoutNullStreams): () => string {
let output = "";
const append = (chunk: Buffer) => {
output = `${output}${chunk.toString("utf8")}`.slice(-64 * 1_024);
};
child.stdout.on("data", append);
child.stderr.on("data", append);
return () => output;
}
async function waitForMockServer(port: number, deadlineAt: number): Promise<void> {
let lastError: unknown;
while (remainingMs(deadlineAt) > 0) {
try {
const result = await requestHttp({
accept: "application/json",
deadlineAt,
path: "/health",
port,
});
if (result.status === 200) {
return;
}
} catch (error) {
lastError = error;
}
await delay(Math.min(25, remainingMs(deadlineAt)));
}
const detail =
lastError instanceof Error
? lastError.message
: typeof lastError === "string"
? lastError
: "timeout";
throw new Error(`mock provider did not become healthy: ${detail}`);
}
async function waitForGatewayDispatchReady(
readOutput: () => string,
deadlineAt: number,
): Promise<void> {
while (remainingMs(deadlineAt) > 0) {
if (readOutput().includes("startup trace: sidecars.ready ")) {
return;
}
await delay(Math.min(25, remainingMs(deadlineAt)));
}
throw new Error("gateway did not finish dispatch-ready sidecars");
}
function buildConfig(root: string, mockPort: number, concurrency: number): string {
const controlUiRoot = path.join(root, "control-ui");
mkdirSync(controlUiRoot, { recursive: true });
copyFileSync(
path.join(process.cwd(), "ui", "index.html"),
path.join(controlUiRoot, "index.html"),
);
const config = structuredClone(BASE_GATEWAY_BENCH_CONFIG) as Record<string, unknown>;
config.gateway = {
...(config.gateway as Record<string, unknown>),
controlUi: { enabled: true, root: controlUiRoot },
};
applyMockOpenAiModelConfig(config, { mockPort, modelRef: "openai/gpt-5.6-luna" });
const agents = config.agents as Record<string, unknown>;
agents.defaults = {
...(agents.defaults as Record<string, unknown>),
maxConcurrent: concurrency,
};
return writeGatewayBenchConfig(root, config, {});
}
async function connectGateway(port: number, deadlineAt: number) {
const client = createGatewayWsClient({
handshakeTimeoutMs: Math.min(8_000, requireRemainingMs(deadlineAt, "connecting WebSocket")),
openTimeoutMs: Math.min(8_000, requireRemainingMs(deadlineAt, "opening WebSocket")),
url: `ws://127.0.0.1:${port}`,
});
await client.waitOpen();
const requestRpc = async <T>(
method: string,
params: unknown,
requestedTimeoutMs?: number,
): Promise<T> => {
const response = await client.request(
method,
params,
Math.max(
1,
Math.min(
requestedTimeoutMs ?? 65_000,
requireRemainingMs(deadlineAt, `waiting for ${method}`),
),
),
);
if (!response.ok) {
const message =
response.error && typeof response.error === "object" && "message" in response.error
? String(response.error.message)
: JSON.stringify(response.error);
throw new Error(`${method} failed: ${message}`);
}
return response.payload as T;
};
await requestRpc("connect", {
minProtocol: PROTOCOL_VERSION,
maxProtocol: PROTOCOL_VERSION,
client: {
id: "gateway-client",
displayName: "gateway-concurrency-benchmark",
version: "1.0.0",
platform: process.platform,
mode: "backend",
},
role: "operator",
scopes: ["operator.read", "operator.write", "operator.admin"],
caps: [],
});
return { close: client.close, request: requestRpc };
}
async function runTurn(rpc: GatewayRpc, index: number, deadlineAt: number): Promise<void> {
const requestedRunId = randomUUID();
const started = await rpc<{ runId?: string; status?: string }>("agent", {
sessionKey: `agent:main:gateway-concurrency-${index + 1}`,
message: `Reply with benchmark stream ${index + 1}.`,
deliver: false,
idempotencyKey: requestedRunId,
});
if (started.status === "ok") {
return;
}
if (started.status !== "accepted") {
throw new Error(`agent ${index + 1} was not accepted: ${JSON.stringify(started)}`);
}
const remaining = requireRemainingMs(deadlineAt, `waiting for agent ${index + 1} completion`);
const waitTimeoutMs = Math.max(
0,
Math.min(60_000, Math.floor(remaining - AGENT_WAIT_RPC_GRACE_MS)),
);
const rpcTimeoutMs = Math.min(65_000, Math.max(1, Math.ceil(remaining)));
const completed = await rpc<{ status?: string }>(
"agent.wait",
{
runId: started.runId ?? requestedRunId,
timeoutMs: waitTimeoutMs,
},
rpcTimeoutMs,
);
if (completed.status !== "ok") {
throw new Error(`agent ${index + 1} did not complete: ${JSON.stringify(completed)}`);
}
}
async function sampleGateway(params: {
deadlineAt: number;
port: number;
rpc: GatewayRpc;
runStartedAt: number;
}): Promise<{ controlUi: TimedProbe; readyz: ReadyProbe; sessionsList: TimedProbe }> {
const atMs = performance.now() - params.runStartedAt;
const safeHttpProbe = async (pathValue: string, accept: string) => {
const startedAt = performance.now();
try {
return {
...(await requestHttp({
accept,
deadlineAt: params.deadlineAt,
path: pathValue,
port: params.port,
})),
ok: true,
};
} catch {
return {
body: "",
latencyMs: performance.now() - startedAt,
ok: false,
status: 0,
};
}
};
const [readyz, controlUi, sessions] = await Promise.all([
safeHttpProbe("/readyz", "application/json"),
safeHttpProbe("/", "text/html"),
(async () => {
const startedAt = performance.now();
try {
const payload = await params.rpc("sessions.list", {}, HTTP_TIMEOUT_MS);
return { latencyMs: performance.now() - startedAt, ok: true, payload };
} catch {
return { latencyMs: performance.now() - startedAt, ok: false, payload: null };
}
})(),
]);
const readyBody = (() => {
if (readyz.status !== 200) {
return {};
}
try {
return JSON.parse(readyz.body) as { eventLoop?: Record<string, unknown> };
} catch {
return {};
}
})();
const eventLoop = readyBody.eventLoop;
return {
controlUi: {
atMs,
latencyMs: controlUi.latencyMs,
ok: controlUi.ok && controlUi.status === 200 && controlUi.body.includes("<html"),
},
readyz: {
atMs,
latencyMs: readyz.latencyMs,
ok: readyz.ok && readyz.status === 200,
status: readyz.status,
degraded: typeof eventLoop?.degraded === "boolean" ? eventLoop.degraded : null,
degradedSinceMs: numberOrNull(eventLoop?.degradedSinceMs),
delayP99Ms: numberOrNull(eventLoop?.delayP99Ms),
utilization: numberOrNull(eventLoop?.utilization),
cpuCoreRatio: numberOrNull(eventLoop?.cpuCoreRatio),
},
sessionsList: { atMs, latencyMs: sessions.latencyMs, ok: sessions.ok },
};
}
async function runGatewaySample(options: {
cadenceMs: number;
concurrency: number;
deadlineAt: number;
entry: string;
}): Promise<BenchmarkRun> {
const root = mkdtempSync(path.join(tmpdir(), "openclaw-gateway-concurrency-"));
const [port, mockPort] = await Promise.all([getFreePort(), getFreePort()]);
const runStartedAt = performance.now();
let gateway: ChildProcessWithoutNullStreams | undefined;
let mockProvider: ChildProcessWithoutNullStreams | undefined;
let client: Awaited<ReturnType<typeof connectGateway>> | undefined;
let gatewayOutput = () => "";
let mockOutput = () => "";
try {
const configPath = buildConfig(root, mockPort, options.concurrency);
mockProvider = spawn(process.execPath, ["scripts/e2e/mock-openai-server.mjs"], {
cwd: process.cwd(),
detached: process.platform !== "win32",
env: {
LANG: process.env.LANG ?? "en_US.UTF-8",
PATH: process.env.PATH,
MOCK_PORT: String(mockPort),
MOCK_RESPONSE_CHUNK_DELAY_MS: String(MOCK_RESPONSE_CHUNK_DELAY_MS),
SUCCESS_MARKER: "OpenClaw gateway concurrency benchmark streaming response.",
},
});
mockOutput = captureChildOutput(mockProvider);
await waitForMockServer(mockPort, options.deadlineAt);
gateway = spawn(process.execPath, buildGatewayBenchChildArgs(options.entry, port), {
cwd: process.cwd(),
detached: process.platform !== "win32",
env: {
...createGatewayBenchEnv(root, configPath, {
caseEnv: { OPENCLAW_SKIP_CHANNELS: "1" },
}),
OPENAI_API_KEY: "gateway-concurrency-benchmark",
},
});
gatewayOutput = captureChildOutput(gateway);
const ready = await waitForInitialProbe({
deadlineAt: options.deadlineAt,
isDone: () => gateway?.exitCode != null || gateway?.signalCode != null,
path: "/readyz",
port,
startAt: runStartedAt,
});
if (ready.status !== 200) {
throw new Error(`gateway did not become ready\n${gatewayOutput()}`);
}
await waitForGatewayDispatchReady(gatewayOutput, options.deadlineAt);
client = await connectGateway(port, options.deadlineAt);
const rpc = client.request;
const baseline = await sampleGateway({
deadlineAt: options.deadlineAt,
port,
rpc,
runStartedAt,
});
if (!baseline.readyz.ok || !baseline.sessionsList.ok || !baseline.controlUi.ok) {
throw new Error("gateway probes did not pass before concurrent load");
}
const controlUi: TimedProbe[] = [];
const readyz: ReadyProbe[] = [];
const sessionsList: TimedProbe[] = [];
let turnsDone = false;
const turnsStartedAt = performance.now();
const turns = Promise.all(
Array.from({ length: options.concurrency }, (_, index) =>
runTurn(rpc, index, options.deadlineAt),
),
).finally(() => {
turnsDone = true;
});
const sampler = (async () => {
for (;;) {
const sampleStartedAt = performance.now();
const sample = await sampleGateway({
deadlineAt: options.deadlineAt,
port,
rpc,
runStartedAt,
});
readyz.push(sample.readyz);
sessionsList.push(sample.sessionsList);
controlUi.push(sample.controlUi);
if (turnsDone || readyz.length >= MAX_SAMPLES_PER_RUN) {
break;
}
await delay(
Math.min(
Math.max(0, options.cadenceMs - (performance.now() - sampleStartedAt)),
requireRemainingMs(options.deadlineAt, "sampling gateway load"),
),
);
}
})();
await Promise.all([turns, sampler]);
const turnsDurationMs = performance.now() - turnsStartedAt;
return {
controlUi,
durationMs: performance.now() - runStartedAt,
readyz,
sessionsList,
turnCount: options.concurrency,
turnsDurationMs,
};
} catch (error) {
const detail = [
error instanceof Error ? error.message : String(error),
gatewayOutput() ? `gateway output:\n${gatewayOutput()}` : "",
mockOutput() ? `mock provider output:\n${mockOutput()}` : "",
]
.filter(Boolean)
.join("\n");
throw new Error(detail, { cause: error });
} finally {
client?.close();
if (gateway) {
await stopChild(gateway);
}
if (mockProvider) {
await stopChild(mockProvider);
}
rmSync(root, { force: true, maxRetries: 3, recursive: true, retryDelay: 100 });
}
}
function summarizeRuns(runs: readonly BenchmarkRun[]) {
const readyz = runs.flatMap((run) => run.readyz);
return {
controlUiFailedSamples: runs.flatMap((run) => run.controlUi).filter((sample) => !sample.ok)
.length,
controlUiLatencyMs: summarizeNumbers(
runs.flatMap((run) => run.controlUi.map((sample) => sample.latencyMs)),
),
cpuCoreRatio: summarizeNumbers(
readyz.flatMap((sample) => (sample.cpuCoreRatio == null ? [] : [sample.cpuCoreRatio])),
),
degradedSamples: readyz.filter((sample) => sample.degraded === true).length,
eventLoopDelayP99Ms: summarizeNumbers(
readyz.flatMap((sample) => (sample.delayP99Ms == null ? [] : [sample.delayP99Ms])),
),
eventLoopUtilization: summarizeNumbers(
readyz.flatMap((sample) => (sample.utilization == null ? [] : [sample.utilization])),
),
readyzLatencyMs: summarizeNumbers(readyz.map((sample) => sample.latencyMs)),
readyzFailedSamples: readyz.filter((sample) => !sample.ok).length,
sampleCount: readyz.length,
sessionsListLatencyMs: summarizeNumbers(
runs.flatMap((run) => run.sessionsList.map((sample) => sample.latencyMs)),
),
sessionsListFailedSamples: runs
.flatMap((run) => run.sessionsList)
.filter((sample) => !sample.ok).length,
turnsDurationMs: summarizeNumbers(runs.map((run) => run.turnsDurationMs)),
};
}
async function main(): Promise<void> {
const argv = process.argv.slice(2);
if (hasHelpFlag(argv)) {
printUsage();
return;
}
const options = parseOptions(argv);
const deadlineAt = performance.now() + options.timeoutMs;
const runs: BenchmarkRun[] = [];
const total = options.runs + options.warmup;
for (let index = 0; index < total; index += 1) {
requireRemainingMs(deadlineAt, "starting gateway run");
const run = await runGatewaySample({ ...options, deadlineAt });
if (index >= options.warmup) {
runs.push(run);
console.error(
`[bench-gateway-concurrency] run ${runs.length}/${options.runs}: turns=${run.turnCount} samples=${run.readyz.length} duration=${run.durationMs.toFixed(1)}ms`,
);
} else {
console.error(
`[bench-gateway-concurrency] warmup ${index + 1}/${options.warmup}: duration=${run.durationMs.toFixed(1)}ms`,
);
}
}
const payload = {
cadenceMs: options.cadenceMs,
concurrency: options.concurrency,
entry: options.entry,
generatedAt: new Date().toISOString(),
mode: "mock-streaming-agent",
runs,
summary: summarizeRuns(runs),
};
if (options.output) {
mkdirSync(path.dirname(options.output), { recursive: true });
writeFileSync(options.output, `${JSON.stringify(payload, null, 2)}\n`);
}
if (options.json || !options.output) {
console.log(JSON.stringify(payload, null, 2));
}
}
export const testing = {
parseOptions,
runTurn,
summarizeNumbers,
summarizeRuns,
};
if (import.meta.url === pathToFileURL(process.argv[1] ?? "").href) {
void main()
.catch((error: unknown) => {
console.error(error instanceof CliArgumentError ? error.message : (error as Error)?.stack);
process.exitCode = 1;
})
.finally(() => {
if (process.exitCode && process.exitCode !== 0) {
console.error(`[bench-gateway-concurrency] FAILED (exit ${process.exitCode})`);
}
});
}
scripts/check-gateway-cpu-scenarios.mjs
+134 -1
View File
@@ -33,6 +33,12 @@ const SINGLE_VALUE_FLAGS = new Set([
]);
const DEFAULT_CPU_CORE_WARN = 0.9;
const DEFAULT_HOT_WALL_WARN_MS = 30_000;
const DEFAULT_GATEWAY_CONCURRENCY = 8;
// Local N=8 mock-stream p99s were 1.3-1.5s on the shared maintainer host;
// 3s leaves roughly 2x headroom while still flagging a material regression.
const CONCURRENCY_EVENT_LOOP_DELAY_P99_WARN_MS = 3_000;
const CONCURRENCY_RPC_P99_WARN_MS = 3_000;
const CONCURRENCY_CONTROL_UI_P99_WARN_MS = 3_000;
const PRIVATE_QA_REQUIRED_DIST_ENTRIES = [
"dist/plugin-sdk/qa-lab.js",
"dist/plugin-sdk/qa-runtime.js",
@@ -191,6 +197,75 @@ function readStartupReport(startupOutput) {
}
}
function validateConcurrencyReport(report) {
if (!report || typeof report !== "object" || Array.isArray(report)) {
return "concurrency report must be a JSON object";
}
if (report.mode !== "mock-streaming-agent") {
return "concurrency report has an unexpected mode";
}
if (!Array.isArray(report.runs) || report.runs.length === 0) {
return "concurrency report has no measured runs";
}
if (!report.summary || typeof report.summary !== "object") {
return "concurrency report missing summary";
}
return null;
}
function readConcurrencyReport(concurrencyOutput) {
if (!fs.existsSync(concurrencyOutput)) {
return {
diagnosticFailure: "concurrency-report-missing",
diagnosticDetail: `expected concurrency bench report at ${concurrencyOutput}`,
report: null,
};
}
try {
const report = readJsonIfExists(concurrencyOutput);
const invalidReason = validateConcurrencyReport(report);
return invalidReason
? {
diagnosticFailure: "concurrency-report-invalid",
diagnosticDetail: invalidReason,
report: null,
}
: { diagnosticFailure: null, diagnosticDetail: null, report };
} catch (error) {
return {
diagnosticFailure: "concurrency-report-invalid",
diagnosticDetail: error instanceof Error ? error.message : String(error),
report: null,
};
}
}
function collectConcurrencyWarnings(report) {
if (!report?.summary) {
return [];
}
const candidates = [
{
kind: "event-loop-delay-p99",
value: report.summary.eventLoopDelayP99Ms?.max,
threshold: CONCURRENCY_EVENT_LOOP_DELAY_P99_WARN_MS,
},
{
kind: "sessions-list-p99",
value: report.summary.sessionsListLatencyMs?.p99,
threshold: CONCURRENCY_RPC_P99_WARN_MS,
},
{
kind: "control-ui-p99",
value: report.summary.controlUiLatencyMs?.p99,
threshold: CONCURRENCY_CONTROL_UI_P99_WARN_MS,
},
];
return candidates.flatMap((candidate) =>
typeof candidate.value === "number" && candidate.value > candidate.threshold ? [candidate] : [],
);
}
function runStep(name, command, args, options = {}, params = {}) {
console.error(`[gateway-cpu] start ${name}`);
const spawn = params.spawnSync ?? defaultSpawnSync;
@@ -287,6 +362,7 @@ async function runGatewayCpuScenarios(options, params = {}) {
fs.mkdirSync(options.outputDir, { recursive: true });
const startupOutput = path.join(options.outputDir, "gateway-startup-bench.json");
const concurrencyOutput = path.join(options.outputDir, "gateway-concurrency-bench.json");
const qaOutputDir = path.join(options.outputDir, "qa-suite");
const qaSummaryPath = path.join(qaOutputDir, "qa-suite-summary.json");
const qaState = options.skipQa ? null : createQaState(options.outputDir);
@@ -329,6 +405,29 @@ async function runGatewayCpuScenarios(options, params = {}) {
)
: { name: "startup bench", signal: null, status: 1 },
);
steps.push(
startupBuild.status === 0
? runStep(
"concurrency bench",
process.execPath,
[
"--import",
"tsx",
"scripts/bench-gateway-concurrency.ts",
"--concurrency",
String(DEFAULT_GATEWAY_CONCURRENCY),
"--runs",
String(options.runs),
"--warmup",
String(options.warmup),
"--output",
concurrencyOutput,
],
{ env: baseEnv },
params,
)
: { name: "concurrency bench", signal: null, status: 1 },
);
}
let privateQaBuildFailed = false;
@@ -373,6 +472,15 @@ async function runGatewayCpuScenarios(options, params = {}) {
? (startupReportResult?.diagnosticFailure ?? null)
: null;
const startup = startupReportResult?.report ?? null;
const concurrencyReportResult = options.skipStartup
? null
: readConcurrencyReport(concurrencyOutput);
const concurrencyReportFailure =
steps.find((step) => step.name === "concurrency bench")?.status === 0
? (concurrencyReportResult?.diagnosticFailure ?? null)
: null;
const concurrency = concurrencyReportResult?.report ?? null;
const concurrencyWarnings = collectConcurrencyWarnings(concurrency);
const qaSummaryResult = options.skipQa ? null : readQaSuiteSummary(qaSummaryPath);
const qaSummaryFailure =
qaStep?.status === 0 ? (qaSummaryResult?.diagnosticFailure ?? null) : null;
@@ -387,6 +495,7 @@ async function runGatewayCpuScenarios(options, params = {}) {
generatedAt: new Date().toISOString(),
outputDir: options.outputDir,
startupOutput: fs.existsSync(startupOutput) ? startupOutput : null,
concurrencyOutput: fs.existsSync(concurrencyOutput) ? concurrencyOutput : null,
qaSummary: fs.existsSync(qaSummaryPath) ? qaSummaryPath : null,
...(startupReportFailure
? {
@@ -400,6 +509,12 @@ async function runGatewayCpuScenarios(options, params = {}) {
qaSummaryFailureDetail: qaSummaryResult?.diagnosticDetail ?? null,
}
: {}),
...(concurrencyReportFailure
? {
concurrencyReportFailure,
concurrencyReportFailureDetail: concurrencyReportResult?.diagnosticDetail ?? null,
}
: {}),
options: {
startupCases: options.startupCases,
qaScenarios: options.qaScenarios,
@@ -407,10 +522,17 @@ async function runGatewayCpuScenarios(options, params = {}) {
warmup: options.warmup,
cpuCoreWarn: options.cpuCoreWarn,
hotWallWarnMs: options.hotWallWarnMs,
concurrency: DEFAULT_GATEWAY_CONCURRENCY,
concurrencyWarnThresholds: {
eventLoopDelayP99Ms: CONCURRENCY_EVENT_LOOP_DELAY_P99_WARN_MS,
sessionsListP99Ms: CONCURRENCY_RPC_P99_WARN_MS,
controlUiP99Ms: CONCURRENCY_CONTROL_UI_P99_WARN_MS,
},
qaStateDir: qaState?.stateDir ?? null,
},
steps,
observations,
concurrencyWarnings,
};
const summaryPath = path.join(options.outputDir, "summary.json");
fs.writeFileSync(summaryPath, `${JSON.stringify(summary, null, 2)}\n`);
@@ -430,12 +552,23 @@ async function runGatewayCpuScenarios(options, params = {}) {
if (startupReportFailure) {
console.error(`[gateway-cpu] fail startup report: ${startupReportResult?.diagnosticDetail}`);
}
for (const warning of concurrencyWarnings) {
console.error(
`[gateway-cpu] warn ${warning.kind}: ${warning.value}ms > ${warning.threshold}ms`,
);
}
if (concurrencyReportFailure) {
console.error(
`[gateway-cpu] fail concurrency report: ${concurrencyReportResult?.diagnosticDetail}`,
);
}
const exitCode =
steps.some((step) => step.status !== 0) ||
observations.length > 0 ||
qaSummaryFailure ||
startupReportFailure
startupReportFailure ||
concurrencyReportFailure
? 1
: 0;
return { exitCode, summary };
src/commands/channels.surfaces-signal-runtime-errors-channels-status-output.test.ts
+2
View File
@@ -94,6 +94,7 @@ describe("channels command", () => {
const lines = formatGatewayChannelsStatusLines({
eventLoop: {
degraded: true,
degradedSinceMs: 180_000,
reasons: ["event_loop_delay", "cpu"],
intervalMs: 62_000,
delayP99Ms: 61_000,
@@ -106,6 +107,7 @@ describe("channels command", () => {
});
expect(lines.join("\n")).toMatch(/Gateway event loop degraded/);
expect(lines.join("\n")).toMatch(/for 3m \(p99 61000ms\)/);
expect(lines.join("\n")).toMatch(/eventLoopDelayMaxMs=62000/);
});
src/commands/channels/status.runtime.ts
+12 -1
View File
@@ -8,6 +8,7 @@ import { formatCliCommand } from "../../cli/command-format.js";
import { getConfiguredChannelsCommandSecretTargetIds } from "../../cli/command-secret-targets.js";
import { readConfigFileSnapshot } from "../../config/config.js";
import { collectChannelStatusIssues } from "../../infra/channels-status-issues.js";
import { formatDurationCompact } from "../../infra/format-time/format-duration.js";
import { formatTimeAgo } from "../../infra/format-time/format-relative.ts";
import { formatPhoneNumberForCli } from "../../infra/phone-number-presentation.js";
import { listConfiguredAnnounceChannelIdsForConfig } from "../../plugins/channel-plugin-ids.js";
@@ -47,7 +48,17 @@ function formatEventLoopBits(value: unknown): string | null {
typeof record.cpuCoreRatio === "number" && Number.isFinite(record.cpuCoreRatio)
? record.cpuCoreRatio
: null;
const degradedSinceMs =
typeof record.degradedSinceMs === "number" && Number.isFinite(record.degradedSinceMs)
? Math.max(0, record.degradedSinceMs)
: null;
const delayP99Ms =
typeof record.delayP99Ms === "number" && Number.isFinite(record.delayP99Ms)
? Math.round(record.delayP99Ms)
: null;
return [
degradedSinceMs != null ? `for ${formatDurationCompact(degradedSinceMs) ?? "0s"}` : null,
delayP99Ms != null ? `(p99 ${delayP99Ms}ms)` : null,
reasons.length ? `reasons=${reasons.join(",")}` : null,
delayMaxMs != null ? `eventLoopDelayMaxMs=${delayMaxMs}` : null,
utilization != null ? `eventLoopUtilization=${utilization}` : null,
@@ -63,7 +74,7 @@ export function formatGatewayChannelsStatusLines(payload: Record<string, unknown
lines.push(theme.success("Gateway reachable."));
const eventLoopLine = formatEventLoopBits(payload.eventLoop);
if (eventLoopLine) {
lines.push(theme.warn(`Gateway event loop degraded: ${eventLoopLine}`));
lines.push(theme.warn(`Gateway event loop degraded ${eventLoopLine}`));
}
const channelLabels =
payload.channelLabels && typeof payload.channelLabels === "object"
src/commands/doctor-whatsapp-responsiveness.test.ts
+6
View File
@@ -102,6 +102,7 @@ describe("doctor WhatsApp responsiveness", () => {
status: {
eventLoop: {
degraded: true,
degradedSinceMs: 61_000,
reasons: ["event_loop_delay"],
intervalMs: 30_000,
delayP99Ms: 42,
@@ -138,6 +139,7 @@ describe("doctor WhatsApp responsiveness", () => {
status: {
eventLoop: {
degraded: true,
degradedSinceMs: 61_000,
reasons: ["event_loop_delay"],
intervalMs: 30_000,
delayP99Ms: 42,
@@ -170,6 +172,7 @@ describe("doctor WhatsApp responsiveness", () => {
status: {
eventLoop: {
degraded: false,
degradedSinceMs: null,
reasons: [],
intervalMs: 1,
delayP99Ms: 0,
@@ -187,6 +190,7 @@ describe("doctor WhatsApp responsiveness", () => {
status: {
eventLoop: {
degraded: true,
degradedSinceMs: 61_000,
reasons: ["event_loop_delay"],
intervalMs: 30_000,
delayP99Ms: 42,
@@ -204,6 +208,7 @@ describe("doctor WhatsApp responsiveness", () => {
status: {
eventLoop: {
degraded: true,
degradedSinceMs: 61_000,
reasons: ["event_loop_delay"],
intervalMs: 30_000,
delayP99Ms: 42,
@@ -228,6 +233,7 @@ describe("doctor WhatsApp responsiveness", () => {
status: {
eventLoop: {
degraded: false,
degradedSinceMs: null,
reasons: [],
intervalMs: 1,
delayP99Ms: 0,
src/commands/health.test.ts
+23
View File
@@ -203,6 +203,29 @@ describe("healthCommand", () => {
expect(output).toContain("Gateway probe duration: 5ms");
});
it("prints persistent event-loop degradation duration in text output", async () => {
const snapshot = {
...createHealthSummary({ channels: {}, channelOrder: [], channelLabels: {} }),
eventLoop: {
degraded: true,
degradedSinceMs: 180_000,
reasons: ["event_loop_delay" as const],
intervalMs: 30_000,
delayP99Ms: 1_200,
delayMaxMs: 1_500,
utilization: 0.75,
cpuCoreRatio: 0.5,
},
};
callGatewayMock.mockResolvedValueOnce(snapshot);
await healthCommand({ json: false, timeoutMs: 1000, config: {} }, runtime as never);
const output = stripAnsi(runtime.log.mock.calls.map((call) => String(call[0])).join("\n"));
expect(output).toContain("Gateway event loop: degraded for 3m");
expect(output).toContain("p99=1200ms");
});
it("omits the probe duration for legacy gateway snapshots", async () => {
const { durationMs, ...legacySnapshot } = createHealthSummary({
channels: {},
src/commands/health.ts
+9 -2
View File
@@ -26,7 +26,10 @@ import type { AgentHealthSummary, HealthSummary } from "../gateway/health/types.
import { info } from "../globals.js";
import { isDiagnosticFlagEnabled } from "../infra/diagnostic-flags.js";
import { formatErrorMessage } from "../infra/errors.js";
import { formatDurationHuman } from "../infra/format-time/format-duration.js";
import {
formatDurationCompact,
formatDurationHuman,
} from "../infra/format-time/format-duration.js";
import { resolveHeartbeatSummaryForAgent } from "../infra/heartbeat-summary.js";
import { createSubsystemLogger } from "../logging/subsystem.js";
import { buildChannelAccountBindings, resolvePreferredAccountId } from "../routing/bindings.js";
@@ -139,8 +142,12 @@ function formatEventLoopHealthLine(summary: HealthSummary): string | null {
return null;
}
const state = eventLoop.degraded ? "degraded" : "ok";
const degradedFor =
eventLoop.degraded && eventLoop.degradedSinceMs != null
? ` for ${formatDurationCompact(eventLoop.degradedSinceMs) ?? "0s"}`
: "";
const reasons = eventLoop.reasons.length > 0 ? ` reasons=${eventLoop.reasons.join(",")}` : "";
return `Gateway event loop: ${state}${reasons} max=${Math.round(
return `Gateway event loop: ${state}${degradedFor}${reasons} max=${Math.round(
eventLoop.delayMaxMs,
)}ms p99=${Math.round(eventLoop.delayP99Ms)}ms util=${eventLoop.utilization} cpu=${
eventLoop.cpuCoreRatio
src/commands/status.command-sections.test.ts
+3 -1
View File
@@ -239,6 +239,7 @@ describe("status.command-sections", () => {
durationMs: 42,
eventLoop: {
degraded: true,
degradedSinceMs: 180_000,
reasons: ["event_loop_delay"],
intervalMs: 62_000,
delayP99Ms: 61_000,
@@ -258,7 +259,8 @@ describe("status.command-sections", () => {
{
Item: "Event loop",
Status: "warn(WARN)",
Detail: "reasons event_loop_delay · max 62000ms · p99 61000ms · util 1 · cpu 1",
Detail:
"degraded for 3m · reasons event_loop_delay · max 62000ms · p99 61000ms · util 1 · cpu 1",
},
]);
});
src/commands/status.command-sections.ts
+5 -1
View File
@@ -9,6 +9,7 @@ import {
} from "../../packages/gateway-protocol/src/connect-error-details.js";
import type { TableColumn } from "../../packages/terminal-core/src/table.js";
import { areRuntimeModelRefsEquivalent } from "../agents/model-runtime-aliases.js";
import { formatDurationCompact } from "../infra/format-time/format-duration.js";
import type { HeartbeatEventPayload } from "../infra/heartbeat-events.js";
import type { Tone } from "../memory-host-sdk/status.js";
import type { SessionStatus, StatusSummary } from "../status/types.js";
@@ -316,13 +317,16 @@ export function buildStatusHealthRows(params: {
/** Formats event-loop latency/utilization health into one table detail string. */
function formatEventLoopHealthDetail(eventLoop: EventLoopHealthLike): string {
const parts = [
eventLoop.degraded && eventLoop.degradedSinceMs != null
? `degraded for ${formatDurationCompact(eventLoop.degradedSinceMs) ?? "0s"}`
: null,
eventLoop.reasons.length > 0 ? `reasons ${eventLoop.reasons.join(",")}` : "healthy",
`max ${Math.round(eventLoop.delayMaxMs)}ms`,
`p99 ${Math.round(eventLoop.delayP99Ms)}ms`,
`util ${eventLoop.utilization}`,
`cpu ${eventLoop.cpuCoreRatio}`,
];
return parts.join(" · ");
return parts.filter((part): part is string => part !== null).join(" · ");
}
/** Builds recent session table rows, optionally including prompt-cache data. */
src/flows/doctor-health-contributions.test.ts
+1
View File
@@ -2326,6 +2326,7 @@ describe("doctor health contributions", () => {
const status = {
eventLoop: {
degraded: true,
degradedSinceMs: 61_000,
reasons: ["event_loop_delay"],
intervalMs: 30_000,
delayP99Ms: 42,
src/gateway/server-lifecycle.ts
+25 -1
View File
@@ -9,7 +9,7 @@ import {
type EffectiveOperatorDeviceIdentity,
} from "../infra/device-pairing.js";
import { upsertPresence } from "../infra/system-presence.js";
import { stopDiagnosticHeartbeat } from "../logging/diagnostic.js";
import { startDiagnosticHeartbeat, stopDiagnosticHeartbeat } from "../logging/diagnostic.js";
import type { createSubsystemLogger } from "../logging/subsystem.js";
import { clearPluginMetadataLifecycleCaches } from "../plugins/plugin-metadata-lifecycle.js";
import { clearSecretsRuntimeSnapshot } from "../secrets/runtime-state.js";
@@ -456,6 +456,30 @@ export async function prepareGatewayLifecycle(params: {
}
};
if (diagnosticsEnabled) {
// Gateway lifecycle owns both this existing heartbeat timer and the monitor
// it samples, so startup failure and normal close tear them down together.
startDiagnosticHeartbeat(undefined, {
getConfig: getRuntimeConfig,
startupGraceMs: 60_000,
sampleLiveness: () => {
const sample = readinessEventLoopHealth.persistentDegradationSnapshot();
if (!sample || sample.degradedSinceMs == null) {
return null;
}
return {
reasons: sample.reasons,
intervalMs: sample.intervalMs,
degradedSinceMs: sample.degradedSinceMs,
eventLoopDelayP99Ms: sample.delayP99Ms,
eventLoopDelayMaxMs: sample.delayMaxMs,
eventLoopUtilization: sample.utilization,
cpuCoreRatio: sample.cpuCoreRatio,
};
},
});
}
return {
...runtime,
completeControlUiDeviceAuthMigrationForEffectiveOperator,
src/gateway/server-methods/channels.status.test.ts
+1
View File
@@ -469,6 +469,7 @@ describe("channelsHandlers channels.status", () => {
});
const eventLoop = {
degraded: true,
degradedSinceMs: 61_000,
reasons: ["event_loop_delay"],
intervalMs: 62_000,
delayP99Ms: 62_000,
src/gateway/server-methods/server-methods.test.ts
+2
View File
@@ -5244,6 +5244,7 @@ describe("gateway healthHandlers.health cache freshness", () => {
it("preserves event-loop health sampled by the refresh path", async () => {
const eventLoop = {
degraded: true,
degradedSinceMs: 61_000,
reasons: ["event_loop_delay" as const],
intervalMs: 2_000,
delayP99Ms: 1_500,
@@ -5253,6 +5254,7 @@ describe("gateway healthHandlers.health cache freshness", () => {
};
const replacementEventLoop = {
degraded: false,
degradedSinceMs: null,
reasons: [],
intervalMs: 1,
delayP99Ms: 0,
src/gateway/server-startup-bootstrap.ts
-8
View File
@@ -9,7 +9,6 @@ import {
} from "../config/config-env-vars.js";
import { assertGatewayConfigEnvSelectionUnchanged } from "../config/gateway-env-selection.js";
import {
getRuntimeConfig,
getRuntimeConfigSourceSnapshot,
readConfigFileSnapshot,
setAppliedRuntimeConfigSnapshot,
@@ -34,7 +33,6 @@ import { isVitestRuntimeEnv, logAcceptedEnvOption } from "../infra/env.js";
import { readGatewayRestartHandoffSync } from "../infra/restart-handoff.js";
import { setGatewaySigusr1RestartPolicy, setPreRestartDeferralCheck } from "../infra/restart.js";
import { enqueueSystemEvent } from "../infra/system-events.js";
import { startDiagnosticHeartbeat } from "../logging/diagnostic.js";
import type { createSubsystemLogger } from "../logging/subsystem.js";
import { setCurrentPluginMetadataSnapshot } from "../plugins/current-plugin-metadata-snapshot.js";
import { getTotalQueueSize } from "../process/command-queue.js";
@@ -338,12 +336,6 @@ export async function prepareGatewayServerBootstrap(input: {
: resolvedStartupAuthOverride;
const diagnosticsEnabled = isDiagnosticsEnabled(cfgAtStart);
setDiagnosticsEnabledForProcess(diagnosticsEnabled);
if (diagnosticsEnabled) {
startDiagnosticHeartbeat(undefined, {
getConfig: getRuntimeConfig,
startupGraceMs: 60_000,
});
}
setGatewaySigusr1RestartPolicy({ allowExternal: isRestartEnabled(cfgAtStart) });
const activeTaskCount = { get: () => 0 };
setPreRestartDeferralCheck(
src/gateway/server/event-loop-health.test.ts
+50
View File
@@ -93,6 +93,7 @@ function expectSnapshotFields(snapshot: unknown, expected: Record<string, unknow
function expectSaturatedLoadSnapshot(snapshot: unknown) {
return expectSnapshotFields(snapshot, {
degraded: true,
degradedSinceMs: 0,
reasons: ["event_loop_utilization", "cpu"],
intervalMs: 1_000,
delayP99Ms: 30,
@@ -114,6 +115,7 @@ describe("createGatewayEventLoopHealthMonitor", () => {
harness.setNow(1_000);
expectSnapshotFields(harness.monitor.snapshot(), {
degraded: false,
degradedSinceMs: null,
reasons: [],
intervalMs: 1_000,
delayP99Ms: 0,
@@ -151,6 +153,7 @@ describe("createGatewayEventLoopHealthMonitor", () => {
expectSnapshotFields(harness.monitor.snapshot(), {
degraded: false,
degradedSinceMs: null,
reasons: [],
intervalMs: 1_000,
utilization: 0.2,
@@ -158,6 +161,53 @@ describe("createGatewayEventLoopHealthMonitor", () => {
});
});
it("tracks continuous degradation and clears it on the first healthy snapshot", () => {
const harness = createMonitorHarness({ cpuMsPerWallMs: 0.1, utilization: 0.2 });
harness.setNow(1_000);
expectSnapshotFields(harness.monitor.snapshot(), {
degraded: false,
degradedSinceMs: null,
});
harness.setDelay({ maxMs: 1_500 });
harness.setNow(2_000);
expectSnapshotFields(harness.monitor.snapshot(), {
degraded: true,
degradedSinceMs: 0,
});
harness.setDelay({ maxMs: 1_500 });
harness.setNow(3_500);
expectSnapshotFields(harness.monitor.snapshot(), {
degraded: true,
degradedSinceMs: 1_500,
});
harness.setNow(4_500);
expectSnapshotFields(harness.monitor.snapshot(), {
degraded: false,
degradedSinceMs: null,
});
});
it("exposes persistent degradation only after the warning threshold", () => {
const harness = createMonitorHarness({ cpuMsPerWallMs: 0.1, utilization: 0.2 });
harness.setDelay({ maxMs: 1_500 });
harness.setNow(1_000);
expect(harness.monitor.persistentDegradationSnapshot()).toBeUndefined();
harness.setDelay({ maxMs: 1_500 });
harness.setNow(60_999);
expect(harness.monitor.persistentDegradationSnapshot()).toBeUndefined();
harness.setDelay({ maxMs: 1_500 });
harness.setNow(61_000);
expectSnapshotFields(harness.monitor.persistentDegradationSnapshot(), {
degraded: true,
degradedSinceMs: 60_000,
});
});
it("keeps rate baselines and the last snapshot until a full sample window is available", () => {
const harness = createMonitorHarness({ cpuMsPerWallMs: 0.1, utilization: 0.2 });
harness.setNow(1_000);
src/gateway/server/event-loop-health.ts
+75 -51
View File
@@ -5,6 +5,7 @@ const EVENT_LOOP_MONITOR_RESOLUTION_MS = 20;
const EVENT_LOOP_DELAY_WARN_MS = 1_000;
const EVENT_LOOP_UTILIZATION_WARN = 0.95;
const CPU_CORE_RATIO_WARN = 0.9;
const PERSISTENT_DEGRADATION_WARN_AFTER_MS = 60_000;
// Load counters can spike during frequent short async wakeups; delay is the blocking signal.
const LOAD_DEGRADATION_DELAY_COEVIDENCE_MS = 25;
const SUSTAINED_LOAD_SAMPLE_MIN_INTERVAL_MS = 1_000;
@@ -17,6 +18,7 @@ type GatewayEventLoopHealthReason = "event_loop_delay" | "event_loop_utilization
export type GatewayEventLoopHealth = {
degraded: boolean;
degradedSinceMs: number | null;
reasons: GatewayEventLoopHealthReason[];
intervalMs: number;
delayP99Ms: number;
@@ -27,6 +29,7 @@ export type GatewayEventLoopHealth = {
type GatewayEventLoopHealthMonitor = {
snapshot: () => GatewayEventLoopHealth | undefined;
persistentDegradationSnapshot: () => GatewayEventLoopHealth | undefined;
stop: () => void;
};
@@ -94,7 +97,7 @@ function classifyGatewayEventLoopHealthReasons(
export function createGatewayEventLoopHealthMonitor(
deps: GatewayEventLoopHealthMonitorDeps = {},
): GatewayEventLoopHealthMonitor {
const nowMs = deps.now ?? Date.now;
const nowMs = deps.now ?? performance.now.bind(performance);
const readCpuUsage = deps.cpuUsage ?? process.cpuUsage.bind(process);
const readEventLoopUtilization =
deps.eventLoopUtilization ?? performance.eventLoopUtilization.bind(performance);
@@ -102,10 +105,11 @@ export function createGatewayEventLoopHealthMonitor(
deps.createDelayMonitor ??
((resolutionMs: number) => monitorEventLoopDelay({ resolution: resolutionMs }));
let monitor: EventLoopDelayMonitor | null = null;
let lastWallAt = nowMs();
let lastWallAt: number | null = nowMs();
let lastCpuUsage: CpuUsage | null = readCpuUsage();
let lastEventLoopUtilization: EventLoopUtilization | null = readEventLoopUtilization();
let lastSnapshot: GatewayEventLoopHealth | undefined;
let firstDegradedAtMs: number | null = null;
try {
monitor = createDelayMonitor(EVENT_LOOP_MONITOR_RESOLUTION_MS);
@@ -115,63 +119,83 @@ export function createGatewayEventLoopHealthMonitor(
monitor = null;
}
const snapshot = (): GatewayEventLoopHealth | undefined => {
if (!monitor || !lastCpuUsage || !lastEventLoopUtilization || lastWallAt === null) {
return undefined;
}
const now = nowMs();
const intervalMs = Math.max(1, now - lastWallAt);
const delayP99Ms = nanosecondsToMilliseconds(monitor.percentile(99));
const delayMaxMs = nanosecondsToMilliseconds(monitor.max);
const hasDelayWarning =
delayP99Ms >= EVENT_LOOP_DELAY_WARN_MS || delayMaxMs >= EVENT_LOOP_DELAY_WARN_MS;
if (!hasDelayWarning && intervalMs < SUSTAINED_LOAD_SAMPLE_MIN_INTERVAL_MS) {
return lastSnapshot;
}
const cpuUsage = readCpuUsage(lastCpuUsage);
const currentEventLoopUtilization = readEventLoopUtilization();
const utilization = roundMetric(
readEventLoopUtilization(currentEventLoopUtilization, lastEventLoopUtilization).utilization,
);
const cpuTotalMs = roundMetric((cpuUsage.user + cpuUsage.system) / 1_000, 1);
const cpuCoreRatio = roundMetric(cpuTotalMs / intervalMs);
const reasons = classifyGatewayEventLoopHealthReasons({
intervalMs,
delayP99Ms,
delayMaxMs,
utilization,
cpuCoreRatio,
});
const degraded = reasons.length > 0;
if (degraded) {
firstDegradedAtMs ??= now;
} else {
firstDegradedAtMs = null;
}
const health: GatewayEventLoopHealth = {
degraded,
degradedSinceMs:
firstDegradedAtMs === null ? null : Math.max(0, Math.round(now - firstDegradedAtMs)),
reasons,
intervalMs,
delayP99Ms,
delayMaxMs,
utilization,
cpuCoreRatio,
};
monitor.reset();
lastWallAt = now;
lastCpuUsage = readCpuUsage();
lastEventLoopUtilization = currentEventLoopUtilization;
lastSnapshot = health;
return health;
};
return {
snapshot: () => {
if (!monitor || !lastCpuUsage || !lastEventLoopUtilization || lastWallAt <= 0) {
return undefined;
}
const now = nowMs();
const intervalMs = Math.max(1, now - lastWallAt);
const delayP99Ms = nanosecondsToMilliseconds(monitor.percentile(99));
const delayMaxMs = nanosecondsToMilliseconds(monitor.max);
const hasDelayWarning =
delayP99Ms >= EVENT_LOOP_DELAY_WARN_MS || delayMaxMs >= EVENT_LOOP_DELAY_WARN_MS;
if (!hasDelayWarning && intervalMs < SUSTAINED_LOAD_SAMPLE_MIN_INTERVAL_MS) {
return lastSnapshot;
}
const cpuUsage = readCpuUsage(lastCpuUsage);
const currentEventLoopUtilization = readEventLoopUtilization();
const utilization = roundMetric(
readEventLoopUtilization(currentEventLoopUtilization, lastEventLoopUtilization).utilization,
);
const cpuTotalMs = roundMetric((cpuUsage.user + cpuUsage.system) / 1_000, 1);
const cpuCoreRatio = roundMetric(cpuTotalMs / intervalMs);
const reasons = classifyGatewayEventLoopHealthReasons({
intervalMs,
delayP99Ms,
delayMaxMs,
utilization,
cpuCoreRatio,
});
const snapshot: GatewayEventLoopHealth = {
degraded: reasons.length > 0,
reasons,
intervalMs,
delayP99Ms,
delayMaxMs,
utilization,
cpuCoreRatio,
};
monitor.reset();
lastWallAt = now;
lastCpuUsage = readCpuUsage();
lastEventLoopUtilization = currentEventLoopUtilization;
lastSnapshot = snapshot;
return snapshot;
snapshot,
// The diagnostic heartbeat is the timer owner. This filtered pull keeps
// persistence policy with the monitor without adding another gateway loop.
persistentDegradationSnapshot: () => {
const current = snapshot();
return current?.degradedSinceMs != null &&
current.degradedSinceMs >= PERSISTENT_DEGRADATION_WARN_AFTER_MS
? current
: undefined;
},
stop: () => {
monitor?.disable();
monitor = null;
lastWallAt = 0;
lastWallAt = null;
lastCpuUsage = null;
lastEventLoopUtilization = null;
lastSnapshot = undefined;
firstDegradedAtMs = null;
},
};
}
src/gateway/server/health-state.test.ts
+1
View File
@@ -222,6 +222,7 @@ describe("refreshGatewayHealthSnapshot", () => {
const healthState = await loadHealthState();
const eventLoop = {
degraded: true,
degradedSinceMs: 61_000,
reasons: ["event_loop_delay" as const],
intervalMs: 2_000,
delayP99Ms: 1_500,
src/gateway/server/readiness.test.ts
+2
View File
@@ -487,6 +487,7 @@ describe("createReadinessChecker", () => {
const { readiness } = createReadinessHarness({
getEventLoopHealth: () => ({
degraded: true,
degradedSinceMs: 61_000,
reasons: ["cpu", "event_loop_utilization"],
intervalMs: 2_000,
delayP99Ms: 42.1,
@@ -500,6 +501,7 @@ describe("createReadinessChecker", () => {
readySnapshot(FIVE_MIN_MS, {
eventLoop: {
degraded: true,
degradedSinceMs: 61_000,
reasons: ["cpu", "event_loop_utilization"],
intervalMs: 2_000,
delayP99Ms: 42.1,
src/infra/diagnostic-events.ts
+1
View File
@@ -421,6 +421,7 @@ export type DiagnosticLivenessWarningEvent = DiagnosticBaseEvent & {
type: "diagnostic.liveness.warning";
reasons: DiagnosticLivenessWarningReason[];
intervalMs: number;
degradedSinceMs?: number;
eventLoopDelayP99Ms?: number;
eventLoopDelayMaxMs?: number;
eventLoopUtilization?: number;
src/logging/diagnostic-stability.ts
+6 -2
View File
@@ -189,7 +189,11 @@ function resolveDiagnosticLivenessRecordLevel(
const hasBlockingWork = event.waiting > 0 || event.queued > 0;
const hasSustainedEventLoopDelay =
(event.eventLoopDelayP99Ms ?? 0) >= LIVENESS_EVENT_LOOP_DELAY_WARN_MS;
return hasBlockingWork || (event.active > 0 && hasSustainedEventLoopDelay) ? "warning" : "info";
return event.degradedSinceMs !== undefined ||
hasBlockingWork ||
(event.active > 0 && hasSustainedEventLoopDelay)
? "warning"
: "info";
}
function sanitizeDiagnosticEvent(event: DiagnosticEventPayload): DiagnosticStabilityEventRecord {
@@ -360,7 +364,7 @@ function sanitizeDiagnosticEvent(event: DiagnosticEventPayload): DiagnosticStabi
break;
case "diagnostic.liveness.warning":
record.level = resolveDiagnosticLivenessRecordLevel(event);
record.durationMs = event.intervalMs;
record.durationMs = event.degradedSinceMs ?? event.intervalMs;
record.count = event.reasons.length;
assignReasonCode(record, event.reasons[0]);
record.eventLoopDelayP99Ms = event.eventLoopDelayP99Ms;
src/logging/diagnostic.test.ts
+41
View File
@@ -2229,6 +2229,46 @@ describe("stuck session diagnostics threshold", () => {
);
});
it("warns and records the full duration for persistent idle event-loop degradation", () => {
const warnSpy = vi.spyOn(diagnosticLogger, "warn").mockImplementation(() => undefined);
const events: DiagnosticEventPayload[] = [];
const unsubscribe = onDiagnosticEvent((event) => events.push(event));
try {
startDiagnosticHeartbeat(
{ diagnostics: { enabled: true } },
{
emitMemorySample: createEmitMemorySampleMock(),
sampleLiveness: () => ({
reasons: ["event_loop_delay"],
intervalMs: 30_000,
degradedSinceMs: 60_000,
eventLoopDelayP99Ms: 1_200,
eventLoopDelayMaxMs: 1_500,
}),
},
);
vi.advanceTimersByTime(30_000);
} finally {
unsubscribe();
}
expectLoggerMessageContaining(warnSpy, "degradedFor=60s");
expect(events.findLast((event) => event.type === "diagnostic.liveness.warning")).toMatchObject({
degradedSinceMs: 60_000,
});
requireMatchingRecord(
getDiagnosticStabilitySnapshot({ limit: 10 }).events,
{
type: "diagnostic.liveness.warning",
level: "warning",
durationMs: 60_000,
},
"persistent liveness stability event",
);
});
it("suppresses liveness warnings during startupGraceMs while still sampling", () => {
const warnSpy = vi.spyOn(diagnosticLogger, "warn").mockImplementation(() => undefined);
const events: string[] = [];
@@ -2530,6 +2570,7 @@ describe("stuck session diagnostics threshold", () => {
sampleLiveness: () => ({
reasons: ["event_loop_delay"],
intervalMs: 30_000,
degradedSinceMs: 60_000,
eventLoopDelayP99Ms: 1_500,
eventLoopDelayMaxMs: 2_000,
}),
src/logging/diagnostic.ts
+18 -3
View File
@@ -111,6 +111,7 @@ type DiagnosticWorkSnapshot = {
type DiagnosticLivenessSample = {
reasons: DiagnosticLivenessWarningReason[];
intervalMs: number;
degradedSinceMs?: number;
eventLoopDelayP99Ms?: number;
eventLoopDelayMaxMs?: number;
eventLoopUtilization?: number;
@@ -427,7 +428,11 @@ function emitDiagnosticLivenessWarning(
const workLabelSummary = formatDiagnosticWorkLabels(work);
const message = `liveness warning: reasons=${sample.reasons.join(",")} interval=${Math.round(
sample.intervalMs / 1000,
)}s eventLoopDelayP99Ms=${formatOptionalDiagnosticMetric(
)}s${
sample.degradedSinceMs === undefined
? ""
: ` degradedFor=${Math.round(sample.degradedSinceMs / 1000)}s`
} eventLoopDelayP99Ms=${formatOptionalDiagnosticMetric(
sample.eventLoopDelayP99Ms,
)} eventLoopDelayMaxMs=${formatOptionalDiagnosticMetric(
sample.eventLoopDelayMaxMs,
@@ -441,9 +446,14 @@ function emitDiagnosticLivenessWarning(
workLabelSummary ? ` work=[${workLabelSummary}]` : ""
}`;
const hasBlockingWork = work.waitingCount > 0 || work.queuedCount > 0;
const hasPersistentDegradation = sample.degradedSinceMs !== undefined;
const hasSustainedEventLoopDelay =
(sample.eventLoopDelayP99Ms ?? 0) >= DEFAULT_LIVENESS_EVENT_LOOP_DELAY_WARN_MS;
if (hasBlockingWork || (hasOpenDiagnosticWork(work) && hasSustainedEventLoopDelay)) {
if (
hasPersistentDegradation ||
hasBlockingWork ||
(hasOpenDiagnosticWork(work) && hasSustainedEventLoopDelay)
) {
diag.warn(message);
} else {
diag.debug(message);
@@ -452,6 +462,7 @@ function emitDiagnosticLivenessWarning(
type: "diagnostic.liveness.warning",
reasons: sample.reasons,
intervalMs: sample.intervalMs,
degradedSinceMs: sample.degradedSinceMs,
eventLoopDelayP99Ms: sample.eventLoopDelayP99Ms,
eventLoopDelayMaxMs: sample.eventLoopDelayMaxMs,
eventLoopUtilization: sample.eventLoopUtilization,
@@ -1194,7 +1205,11 @@ export function startDiagnosticHeartbeat(
if (heartbeatInterval) {
return;
}
startDiagnosticLivenessSampler();
// Gateway supplies its lifecycle-owned monitor; other runtimes retain the
// built-in sampler. Never allocate two perf monitors for one heartbeat.
if (!opts?.sampleLiveness) {
startDiagnosticLivenessSampler();
}
const livenessGraceUntil =
opts?.startupGraceMs != null && opts.startupGraceMs > 0 ? Date.now() + opts.startupGraceMs : 0;
lastDiagnosticHeartbeatTickAt = Date.now();
test/scripts/bench-gateway-concurrency.test.ts
+89
View File
@@ -0,0 +1,89 @@
// Gateway concurrency benchmark tests cover CLI parsing and bounded percentile summaries.
import { spawnSync } from "node:child_process";
import { performance } from "node:perf_hooks";
import { describe, expect, it } from "vitest";
import { testing } from "../../scripts/bench-gateway-concurrency.ts";
describe("gateway concurrency benchmark script", () => {
it("parses benchmark controls without booting a gateway", () => {
expect(
testing.parseOptions([
"--concurrency",
"12",
"--runs",
"2",
"--warmup",
"0",
"--cadence-ms",
"50",
"--timeout-ms",
"90000",
"--output",
"concurrency.json",
"--json",
]),
).toMatchObject({
cadenceMs: 50,
concurrency: 12,
json: true,
output: "concurrency.json",
runs: 2,
timeoutMs: 90_000,
warmup: 0,
});
expect(() => testing.parseOptions(["--concurrency", "65"])).toThrow(
"--concurrency must be at most 64",
);
expect(() => testing.parseOptions(["--runs", "2", "--runs", "3"])).toThrow(
"--runs was provided more than once",
);
expect(() => testing.parseOptions(["--wat"])).toThrow("Unknown argument: --wat");
});
it("reports p50, p95, p99, and max with nearest-rank percentiles", () => {
expect(testing.summarizeNumbers([100, 1, 4, 2, 3])).toEqual({
count: 5,
max: 100,
p50: 3,
p95: 100,
p99: 100,
});
expect(testing.summarizeNumbers([])).toBeNull();
});
it("bounds an accepted turn wait by the benchmark deadline", async () => {
const calls: Array<{ method: string; params: unknown; timeoutMs?: number }> = [];
const rpc = async <T>(method: string, params: unknown, timeoutMs?: number): Promise<T> => {
calls.push({ method, params, timeoutMs });
return (
method === "agent" ? { runId: "run-1", status: "accepted" } : { status: "timeout" }
) as T;
};
await expect(testing.runTurn(rpc, 0, performance.now() + 2_000)).rejects.toThrow(
"agent 1 did not complete",
);
const wait = calls.find((call) => call.method === "agent.wait");
expect(wait?.params).toMatchObject({ runId: "run-1" });
const serverTimeoutMs = (wait?.params as { timeoutMs?: unknown }).timeoutMs;
expect(serverTimeoutMs).toBe(0);
expect(wait?.timeoutMs).toEqual(expect.any(Number));
expect(Number.isInteger(wait?.timeoutMs)).toBe(true);
expect(wait?.timeoutMs).toBeGreaterThan(serverTimeoutMs as number);
expect(wait?.timeoutMs).toBeLessThanOrEqual(2_000);
});
it("ends CLI failures with the required wrapper marker", () => {
const result = spawnSync(
process.execPath,
["--import", "tsx", "scripts/bench-gateway-concurrency.ts", "--wat"],
{ cwd: process.cwd(), encoding: "utf8" },
);
expect(result.status).toBe(1);
expect(result.stderr.trim().split("\n").at(-1)).toBe(
"[bench-gateway-concurrency] FAILED (exit 1)",
);
});
});
test/scripts/check-gateway-cpu-scenarios.test.ts
+23
View File
@@ -50,6 +50,21 @@ function writeQaSuiteSummary(
);
}
function writeConcurrencyReport(outputDir: string): void {
writeFileSync(
path.join(outputDir, "gateway-concurrency-bench.json"),
`${JSON.stringify({
mode: "mock-streaming-agent",
runs: [{ turnCount: 8 }],
summary: {
eventLoopDelayP99Ms: { max: 20 },
sessionsListLatencyMs: { p99: 30 },
controlUiLatencyMs: { p99: 25 },
},
})}\n`,
);
}
afterEach(() => {
for (const root of tempRoots.splice(0)) {
rmSync(root, { recursive: true, force: true });
@@ -160,6 +175,9 @@ describe("gateway CPU scenario guard", () => {
if (args.includes("scripts/bench-gateway-startup.ts")) {
writeFileSync(startupOutput, `${JSON.stringify({ results: [{ id: "default" }] })}\n`);
}
if (args.includes("scripts/bench-gateway-concurrency.ts")) {
writeConcurrencyReport(outputDir);
}
return { status: 0 };
},
});
@@ -168,10 +186,13 @@ describe("gateway CPU scenario guard", () => {
expect(calls.map((call) => call.args[0])).toEqual([
"scripts/ensure-cli-startup-build.mjs",
"--import",
"--import",
]);
expect(calls[1]?.args).toContain("scripts/bench-gateway-startup.ts");
expect(calls[2]?.args).toContain("scripts/bench-gateway-concurrency.ts");
expect(calls[0]?.env?.PNPM_CONFIG_VERIFY_DEPS_BEFORE_RUN).toBe("false");
expect(calls[1]?.env?.PNPM_CONFIG_VERIFY_DEPS_BEFORE_RUN).toBe("false");
expect(calls[2]?.env?.PNPM_CONFIG_VERIFY_DEPS_BEFORE_RUN).toBe("false");
});
it("fails successful startup benches that do not write a report", async () => {
@@ -248,6 +269,7 @@ describe("gateway CPU scenario guard", () => {
expect(result.summary.steps).toEqual([
{ name: "startup build", signal: null, status: 1 },
{ name: "startup bench", signal: null, status: 1 },
{ name: "concurrency bench", signal: null, status: 1 },
]);
});
@@ -273,6 +295,7 @@ describe("gateway CPU scenario guard", () => {
expect(result.summary.steps).toEqual([
{ name: "startup build", error: "spawn ENOENT", signal: null, status: 1 },
{ name: "startup bench", signal: null, status: 1 },
{ name: "concurrency bench", signal: null, status: 1 },
]);
});