diff --git a/config/knip.config.ts b/config/knip.config.ts index 86a59c1c3036..7683e62f8715 100644 --- a/config/knip.config.ts +++ b/config/knip.config.ts @@ -117,6 +117,8 @@ const rootEntries = [ "scripts/print-cli-backend-live-metadata.ts!", // Workflow/package-script entrypoints are not imported from production modules. "scripts/openclaw-cross-os-release-checks.ts!", + // Spawned by the agent concurrency benchmark; no static import edge exists. + "scripts/bench-agent-concurrency-worker.ts!", "scripts/bench-sqlite-reliability.ts!", // Docker/manual E2E executables and their nested assertion/probe entrypoints. "scripts/e2e/*.{js,mjs,ts}!", diff --git a/scripts/bench-agent-concurrency-worker.ts b/scripts/bench-agent-concurrency-worker.ts new file mode 100644 index 000000000000..be424b2edc73 --- /dev/null +++ b/scripts/bench-agent-concurrency-worker.ts @@ -0,0 +1,810 @@ +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { performance } from "node:perf_hooks"; +import { pathToFileURL } from "node:url"; +import type { SubagentRunRecord } from "../src/agents/subagent-registry.types.js"; +import type { DB as OpenClawStateKyselyDatabase } from "../src/state/openclaw-state-db.generated.js"; +import { + WORKER_RESULT_SENTINEL, + type WorkerResult, + type WorkerScenario, +} from "./bench-agent-concurrency.js"; + +type WorkerOptions = { + scenario: WorkerScenario; + size: number; + runs: number; + warmup: number; +}; + +type Sample = { + durationMs: number; + invariant: Record; +}; + +const SCENARIOS = new Set([ + "spawnPipelineInMemory", + "spawnPipelineDurable", + "admission", + "recoverySweep", + "duplicateSuppression", +]); + +function parseInteger(raw: string | undefined, flag: string, min: number, max: number): number { + if (!raw || !/^\d+$/u.test(raw)) { + throw new Error(`${flag} must be an integer`); + } + const value = Number(raw); + if (value < min || value > max) { + throw new Error(`${flag} must be between ${min} and ${max}`); + } + return value; +} + +function parseOptions(argv: string[]): WorkerOptions { + const values = new Map(); + for (let index = 0; index < argv.length; index += 2) { + const flag = argv[index]; + const value = argv[index + 1]; + if (!flag?.startsWith("--") || !value || value.startsWith("--")) { + throw new Error(`invalid worker argument near ${flag ?? "end"}`); + } + if (values.has(flag)) { + throw new Error(`${flag} was provided more than once`); + } + values.set(flag, value); + } + const scenario = values.get("--scenario") as WorkerScenario | undefined; + if (!scenario || !SCENARIOS.has(scenario)) { + throw new Error(`unknown worker scenario: ${scenario ?? "missing"}`); + } + return { + scenario, + size: parseInteger(values.get("--size"), "--size", 1, 4096), + runs: parseInteger(values.get("--runs"), "--runs", 1, 100), + warmup: parseInteger(values.get("--warmup"), "--warmup", 0, 20), + }; +} + +function processMaxRssBytes(): number { + return Math.max(0, Math.round(process.resourceUsage().maxRSS * 1024)); +} + +function toError(error: unknown): Error { + return error instanceof Error ? error : new Error(String(error)); +} + +async function waitForCondition(check: () => boolean): Promise { + const deadline = Date.now() + 30_000; + while (Date.now() < deadline) { + if (check()) { + return true; + } + await new Promise((resolve) => { + setImmediate(resolve); + }); + } + return check(); +} + +async function resetRuntime(persist: boolean): Promise { + const [subagents, tasks, stateDb, agentDb] = await Promise.all([ + import("../src/agents/subagent-registry.test-helpers.js"), + import("../src/tasks/task-runtime.test-helpers.js"), + import("../src/state/openclaw-state-db.js"), + import("../src/state/openclaw-agent-db.js"), + ]); + subagents.resetSubagentRegistryForTests({ persist }); + subagents.testing.setDepsForTest(); + tasks.resetTaskRegistryControlRuntimeForTests(); + tasks.resetTaskRegistryDeliveryRuntimeForTests(); + tasks.resetDetachedTaskLifecycleRuntimeForTests(); + tasks.resetTaskRegistryForTests({ persist }); + tasks.resetTaskFlowRegistryForTests({ persist }); + stateDb.closeOpenClawStateDatabaseForTest(); + agentDb.closeOpenClawAgentDatabasesForTest(); +} + +function createTerminalWaitBarrier() { + const waiters = new Map void>(); + let outstanding = 0; + let releasesStarted = false; + const callGateway = (async (request: { method?: string; params?: unknown }) => { + if (request.method !== "agent.wait") { + return {} as T; + } + if (releasesStarted) { + throw new Error("agent.wait registered after the benchmark barrier released"); + } + const runId = + request.params && + typeof request.params === "object" && + !Array.isArray(request.params) && + typeof (request.params as { runId?: unknown }).runId === "string" + ? (request.params as { runId: string }).runId + : undefined; + if (!runId || waiters.has(runId)) { + throw new Error(`invalid or duplicate benchmark agent.wait run id: ${runId ?? "missing"}`); + } + outstanding += 1; + return await new Promise((resolve) => { + waiters.set(runId, (value) => { + outstanding -= 1; + resolve(value as T); + }); + }); + }) as typeof import("../src/gateway/call.js").callGateway; + return { + callGateway, + get outstanding() { + return outstanding; + }, + async waitUntilBlocked(expected: number) { + return await waitForCondition(() => outstanding === expected); + }, + release(runId: string) { + releasesStarted = true; + const resolve = waiters.get(runId); + if (!resolve) { + throw new Error(`missing blocked benchmark agent.wait for ${runId}`); + } + waiters.delete(runId); + const endedAt = Date.now(); + resolve({ + status: "ok", + startedAt: endedAt - 1, + endedAt, + stopReason: "stop", + }); + }, + releaseAll() { + releasesStarted = true; + for (const runId of waiters.keys()) { + this.release(runId); + } + }, + }; +} + +async function configureSpawnRuntime( + mode: "memory" | "durable", + callGateway: typeof import("../src/gateway/call.js").callGateway, +): Promise { + const [subagents, registry, taskStore, flowStore] = await Promise.all([ + import("../src/agents/subagent-registry.test-helpers.js"), + import("../src/agents/subagent-registry-memory.js"), + import("../src/tasks/task-registry.store.js"), + import("../src/tasks/task-flow-registry.store.test-support.js"), + ]); + const sharedDeps = { + callGateway, + getRuntimeConfig: () => ({}), + onAgentEvent: () => () => {}, + resolveAgentTimeoutMs: () => 1_000, + captureSubagentCompletionReply: async (childSessionKey: string) => { + const entry = [...registry.subagentRuns.values()].find( + (candidate) => candidate.childSessionKey === childSessionKey, + ); + if (entry) { + // Completion already owns the row at this awaited seam. Suppress only + // its unrelated session projection, not the terminal registry/task transition. + entry.execution.suppressSessionEffects = true; + } + return undefined; + }, + cleanupBrowserSessionsForLifecycleEnd: async () => {}, + runSubagentAnnounceFlow: async () => false, + maybeWakeRequesterAfterAllChildrenSettled: async () => false, + ensureContextEnginesInitialized: () => {}, + loadAgentRuntimePluginRegistryHandle: () => undefined, + resolveContextEngine: async () => + ({ + info: { id: "bench", name: "bench", version: "1" }, + ingest: async () => ({ ok: true }), + assemble: async () => ({ messages: [] }), + onSubagentEnded: async () => {}, + }) as unknown as import("../src/context-engine/types.js").ContextEngine, + }; + if (mode === "memory") { + subagents.testing.setDepsForTest({ + ...sharedDeps, + persistSubagentRunsToDisk: () => {}, + persistSubagentRunsToDiskOrThrow: () => {}, + }); + taskStore.configureTaskRegistryRuntime({ + store: { + loadSnapshot: () => ({ tasks: new Map(), deliveryStates: new Map() }), + saveSnapshot: () => {}, + upsertTaskWithDeliveryState: () => {}, + upsertTask: () => {}, + deleteTaskWithDeliveryState: () => {}, + deleteTask: () => {}, + upsertDeliveryState: () => {}, + deleteDeliveryState: () => {}, + close: () => {}, + }, + }); + flowStore.configureTaskFlowRegistryRuntime({ + store: { + loadSnapshot: () => ({ flows: new Map() }), + saveSnapshot: () => {}, + upsertFlow: () => {}, + deleteFlow: () => {}, + close: () => {}, + }, + }); + return; + } + subagents.testing.setDepsForTest(sharedDeps); +} + +type BenchmarkStateDatabase = Pick; + +async function readDurableRows() { + const [{ executeSqliteQuerySync, getNodeSqliteKysely }, stateDb] = await Promise.all([ + import("../src/infra/kysely-sync.js"), + import("../src/state/openclaw-state-db.js"), + ]); + const database = stateDb.openOpenClawStateDatabase(); + const db = getNodeSqliteKysely(database.db); + return { + path: database.path, + subagentRows: executeSqliteQuerySync( + database.db, + db.selectFrom("subagent_runs").select(["run_id", "ended_at"]).orderBy("run_id"), + ).rows, + taskRows: executeSqliteQuerySync( + database.db, + db + .selectFrom("task_runs") + .select(["run_id", "status", "ended_at"]) + .where("runtime", "=", "subagent") + .orderBy("run_id"), + ).rows, + }; +} + +function listSqliteFiles(root: string): string[] { + return fs + .readdirSync(root, { recursive: true }) + .map(String) + .filter((entry) => entry.includes(".sqlite")) + .toSorted(); +} + +async function listBenchmarkTasks() { + const tasks = await import("../src/tasks/task-registry.js"); + return tasks.listTaskRecords().filter((task) => task.runtime === "subagent"); +} + +async function listBenchmarkTaskMemory() { + const state = await import("../src/tasks/task-registry-state.js"); + return [...state.tasks.values()].filter((task) => task.runtime === "subagent"); +} + +function assertExactRunIds(actual: Array, expected: string[], label: string): void { + const normalized = actual + .filter((value): value is string => typeof value === "string") + .toSorted(); + const wanted = expected.toSorted(); + if (JSON.stringify(normalized) !== JSON.stringify(wanted)) { + throw new Error( + `${label} mismatch: ${JSON.stringify({ actual: normalized, expected: wanted })}`, + ); + } +} + +async function runSpawnSample( + fanout: number, + serial: number, + mode: "memory" | "durable", + stateDir: string, +): Promise { + const [pipeline, registry] = await Promise.all([ + import("../src/agents/spawn-pipeline.js"), + import("../src/agents/subagent-registry-memory.js"), + ]); + await resetRuntime(mode === "durable"); + const barrier = createTerminalWaitBarrier(); + await configureSpawnRuntime(mode, barrier.callGateway); + const taskRegistry = await import("../src/tasks/task-registry.js"); + let releases = 0; + const runIds: string[] = []; + const params = Array.from({ length: fanout }, (_, index) => { + const runId = `bench-${mode}-${serial}-${index}`; + runIds.push(runId); + let released = false; + return { + adapter: { + initialize: async () => ({ index }), + dispatchTurn: async () => ({ runId }), + cleanupOnFailure: async () => {}, + }, + admissionReservation: { + release: () => { + if (!released) { + released = true; + releases += 1; + } + }, + }, + buildRegistration: () => ({ + runId, + childSessionKey: `agent:bench:subagent:${mode}:${serial}:${index}`, + requesterSessionKey: "agent:bench:main", + requesterDisplayKey: "bench", + task: `benchmark child ${index}`, + cleanup: "keep" as const, + expectsCompletionMessage: false, + }), + progressSessionKey: "agent:bench:main", + }; + }); + let result: Sample | undefined; + let failure: unknown; + try { + const startedAt = performance.now(); + const pipelineResults = await Promise.all( + params.map((entry) => pipeline.runSpawnPipeline(entry)), + ); + const durationMs = performance.now() - startedAt; + const blocked = await barrier.waitUntilBlocked(fanout); + const successful = pipelineResults.filter((pipelineResult) => pipelineResult.ok); + const uniqueRuns = new Set(successful.map((pipelineResult) => pipelineResult.runId)).size; + const registeredRuns = registry.subagentRuns.size; + const tasksWhileBlocked = await listBenchmarkTasks(); + assertExactRunIds( + tasksWhileBlocked.map((task) => task.runId ?? null), + runIds, + `${mode} in-memory task rows`, + ); + if (!blocked || barrier.outstanding !== fanout) { + throw new Error(`spawn ${mode} did not block ${fanout} agent.wait calls`); + } + if ( + successful.length !== fanout || + uniqueRuns !== fanout || + registeredRuns !== fanout || + releases !== fanout + ) { + throw new Error( + `spawn ${mode} registration invariant failed: ${JSON.stringify({ fanout, uniqueRuns, registeredRuns, releases })}`, + ); + } + let durableSubagentRows = 0; + let durableTaskRows = 0; + let durableStateFile = listSqliteFiles(stateDir).length > 0; + if (mode === "durable") { + const durable = await readDurableRows(); + assertExactRunIds( + durable.subagentRows.map((row) => row.run_id), + runIds, + "durable subagent rows", + ); + assertExactRunIds( + durable.taskRows.map((row) => row.run_id), + runIds, + "durable task rows", + ); + if ( + durable.subagentRows.some((row) => row.ended_at !== null) || + durable.taskRows.some((row) => row.status !== "running" || row.ended_at !== null) + ) { + throw new Error("durable rows settled before the benchmark barrier released"); + } + durableSubagentRows = durable.subagentRows.length; + durableTaskRows = durable.taskRows.length; + durableStateFile = fs.existsSync(durable.path); + if (!durableStateFile) { + throw new Error(`durable spawn pipeline database is missing at ${durable.path}`); + } + } else if (durableStateFile) { + throw new Error("in-memory spawn pipeline created durable SQLite state"); + } + for (const runId of runIds) { + barrier.release(runId); + const runSettled = await waitForCondition(() => { + const run = registry.subagentRuns.get(runId); + const task = taskRegistry.findTaskByRunId(runId); + return ( + run?.execution.status === "terminal" && + typeof run.execution.endedAt === "number" && + typeof run.cleanupCompletedAt === "number" && + task?.status === "succeeded" + ); + }); + if (!runSettled) { + throw new Error(`spawn ${mode} did not settle released run ${runId}`); + } + } + const settledTasks = await listBenchmarkTasks(); + const settledRuns = [...registry.subagentRuns.values()].filter( + (entry) => + entry.execution.status === "terminal" && + typeof entry.execution.endedAt === "number" && + typeof entry.cleanupCompletedAt === "number", + ).length; + const succeededTasks = settledTasks.filter((task) => task.status === "succeeded").length; + if (settledRuns !== fanout || succeededTasks !== fanout || barrier.outstanding !== 0) { + throw new Error( + `spawn ${mode} settlement invariant failed: ${JSON.stringify({ fanout, settledRuns, succeededTasks, outstandingWaits: barrier.outstanding })}`, + ); + } + result = { + durationMs, + invariant: { + ok: true, + registeredRuns, + reservationsReleased: releases, + blockedWaits: fanout, + settledRuns, + settledTasks: succeededTasks, + outstandingWaits: barrier.outstanding, + durableSubagentRows, + durableTaskRows, + durableStateFile, + postTeardownRegistryRows: -1, + postTeardownTaskRows: -1, + postTeardownDurableSubagentRows: -1, + postTeardownDurableTaskRows: -1, + }, + }; + } catch (error) { + failure = error; + } finally { + try { + barrier.releaseAll(); + } catch (error) { + failure ??= error; + } + try { + await resetRuntime(mode === "durable"); + } catch (error) { + failure ??= error; + } + } + if (failure) { + throw toError(failure); + } + const postTeardownTasks = await listBenchmarkTaskMemory(); + const postTeardownRegistryRows = registry.subagentRuns.size; + const postTeardownTaskRows = postTeardownTasks.length; + let postTeardownDurableSubagentRows = 0; + let postTeardownDurableTaskRows = 0; + if (mode === "durable") { + const durable = await readDurableRows(); + postTeardownDurableSubagentRows = durable.subagentRows.length; + postTeardownDurableTaskRows = durable.taskRows.length; + await resetRuntime(false); + } + if ( + postTeardownRegistryRows !== 0 || + postTeardownTaskRows !== 0 || + postTeardownDurableSubagentRows !== 0 || + postTeardownDurableTaskRows !== 0 || + barrier.outstanding !== 0 + ) { + throw new Error( + `spawn ${mode} teardown invariant failed: ${JSON.stringify({ postTeardownRegistryRows, postTeardownTaskRows, postTeardownDurableSubagentRows, postTeardownDurableTaskRows, outstandingWaits: barrier.outstanding })}`, + ); + } + if (mode === "memory" && listSqliteFiles(stateDir).length > 0) { + throw new Error("in-memory spawn pipeline created SQLite state during teardown"); + } + if (!result) { + throw new Error(`spawn ${mode} did not produce a benchmark result`); + } + result.invariant.postTeardownRegistryRows = postTeardownRegistryRows; + result.invariant.postTeardownTaskRows = postTeardownTaskRows; + result.invariant.postTeardownDurableSubagentRows = postTeardownDurableSubagentRows; + result.invariant.postTeardownDurableTaskRows = postTeardownDurableTaskRows; + return result; +} + +async function runAdmissionSample(fanout: number, serial: number): Promise { + const { reserveChildAdmissionSlot } = await import("../src/agents/child-admission.js"); + const controllerSessionKey = `agent:bench:admission:${serial}`; + const reservations: Array<{ release: () => void }> = []; + const reserve = () => + reserveChildAdmissionSlot({ + controllerSessionKey, + resolveAdmission: (pendingChildren) => + pendingChildren < fanout + ? { ok: true as const } + : { ok: false as const, governingCap: "benchmark" }, + }); + const startedAt = performance.now(); + for (let index = 0; index < fanout; index += 1) { + const reservation = reserve(); + if (!reservation.ok) { + throw new Error(`admission rejected slot ${index + 1}/${fanout}`); + } + reservations.push(reservation); + } + const overflow = reserve(); + for (const reservation of reservations) { + reservation.release(); + reservation.release(); + } + const replacement = reserve(); + if (replacement.ok) { + replacement.release(); + } + const durationMs = performance.now() - startedAt; + const ok = !overflow.ok && replacement.ok; + if (!ok) { + throw new Error(`admission invariant failed at fanout ${fanout}`); + } + return { + durationMs, + invariant: { ok, admissionCap: reservations.length, overflowRejected: true, released: true }, + }; +} + +function sweepRow(child: number, generation: number, now: number): SubagentRunRecord { + const current = generation === 3; + return { + runId: `bench-sweep-${child}-${generation}`, + childSessionKey: `agent:bench:subagent:sweep-${child}`, + requesterSessionKey: "agent:bench:main", + requesterDisplayKey: "bench", + task: `sweep child ${child}`, + cleanup: current ? "keep" : "delete", + generation, + createdAt: now - generation, + archiveAtMs: current ? undefined : now - 1, + terminalOwner: current ? "interrupted-recovery" : undefined, + endedReason: current ? "subagent-error" : undefined, + execution: current + ? { + status: "terminal", + startedAt: now - 2_000, + endedAt: now - 1_000, + outcome: { status: "error", error: "interrupted recovery replay" }, + suppressSessionEffects: true, + } + : { + status: "terminal", + startedAt: now - 2_000, + endedAt: now - 1_000, + outcome: { status: "error", error: "retired recovery generation" }, + suppressSessionEffects: true, + }, + }; +} + +async function runSweepSample(childCount: number): Promise { + const { createSubagentRegistrySweeper } = + await import("../src/agents/subagent-registry-sweeper.js"); + const now = Date.now(); + const runs = new Map(); + for (let child = 0; child < childCount; child += 1) { + for (const generation of [3, 2, 1]) { + const entry = sweepRow(child, generation, now); + runs.set(entry.runId, entry); + } + } + let sessionEffects = 0; + let recoveryProjections = 0; + let lostContextCompletions = 0; + const sweeper = createSubagentRegistrySweeper({ + runs, + resumedRuns: new Set(), + persist: () => {}, + clearPendingLifecycleError: () => {}, + clearPendingLifecycleTimeout: () => {}, + sweepPendingLifecycle: () => {}, + completeSubagentRunWithRecovery: async () => { + lostContextCompletions += 1; + }, + getGatewayRecoveryRuntime: () => undefined, + abandonSubagentRestartRecoveryLaunch: () => true, + clearAcceptedSubagentRestartRecovery: () => true, + resumeSettledSubagentRestartRecovery: () => true, + replaceSubagentRunAfterSteer: () => true, + markSubagentRestartRecoveryLaunchAttempted: () => undefined, + markSubagentRestartRecoveryLaunchAccepted: () => undefined, + markSubagentRestartRecoveryLaunchConsumed: () => undefined, + reserveSubagentRestartRecoveryLaunch: () => undefined, + resetSubagentRestartRecoveryLaunchAttempt: () => true, + finalizeInterruptedSubagentRun: async ({ runId, expectedEntry }) => { + if (runs.get(runId) !== expectedEntry || expectedEntry?.generation !== 3) { + throw new Error(`unexpected recovery projection owner: ${runId}`); + } + recoveryProjections += 1; + return 1; + }, + resumeRequesterSettleWake: () => {}, + startSubagentAnnounceCleanupFlow: () => true, + completeCleanupBookkeeping: () => {}, + shouldEmitEndedHookForRun: () => false, + emitSubagentEndedHookForRun: async () => {}, + callGateway: (async () => { + sessionEffects += 1; + return {} as T; + }) as typeof import("../src/gateway/call.js").callGateway, + cleanupCollectorLaunchResources: async () => true, + runContextEngineSubagentEnded: async () => { + sessionEffects += 1; + }, + notifyContextEngineSubagentEnded: async () => { + sessionEffects += 1; + }, + retireSupersededRun: async () => {}, + getRunsForChildSession: (childSessionKey) => + [...runs.values()].filter((entry) => entry.childSessionKey === childSessionKey), + getRunsForCollectorGroup: () => [], + warn: () => {}, + }); + try { + const startedAt = performance.now(); + await sweeper.sweepOnce(); + const durationMs = performance.now() - startedAt; + const retainedCurrent = [...runs.values()].filter((entry) => entry.generation === 3).length; + const removedRows = childCount * 3 - runs.size; + const ok = + removedRows === childCount * 2 && + retainedCurrent === childCount && + sessionEffects === 0 && + recoveryProjections === childCount && + lostContextCompletions === 0; + if (!ok) { + throw new Error( + `registry sweep invariant failed: ${JSON.stringify({ childCount, removedRows, retainedCurrent, sessionEffects, recoveryProjections, lostContextCompletions })}`, + ); + } + return { + durationMs, + invariant: { + ok, + seededRows: childCount * 3, + removedRows, + retainedCurrent, + sessionEffects, + recoveryProjections, + lostContextCompletions, + }, + }; + } finally { + sweeper.reset(); + } +} + +async function runDedupeSample(childCount: number): Promise { + const { dedupeLatestChildCompletionRows } = + await import("../src/agents/subagent-announce-output.js"); + const rowsForOrder = (generations: number[]) => + Array.from({ length: childCount }, (_, child) => + generations.map((generation) => ({ + runId: `bench-dedupe-${child}-${generation}`, + childSessionKey: `agent:bench:subagent:dedupe-${child}`, + task: `dedupe child ${child}`, + generation, + createdAt: generation, + execution: { status: "terminal" as const, endedAt: generation }, + })), + ).flat(); + const newestFirstRows = rowsForOrder([3, 2, 1]); + const oldestFirstRows = rowsForOrder([1, 2, 3]); + const startedAt = performance.now(); + const newestFirst = dedupeLatestChildCompletionRows(newestFirstRows); + const oldestFirst = dedupeLatestChildCompletionRows(oldestFirstRows); + const durationMs = performance.now() - startedAt; + const newestFirstSelectedNewest = newestFirst.every((row) => row.generation === 3); + const oldestFirstSelectedNewest = oldestFirst.every((row) => row.generation === 3); + const ok = + newestFirst.length === childCount && + oldestFirst.length === childCount && + newestFirstSelectedNewest && + oldestFirstSelectedNewest; + if (!ok) { + throw new Error(`completion dedupe invariant failed for ${childCount} children`); + } + return { + durationMs, + invariant: { + ok, + inputRowsPerOrdering: newestFirstRows.length, + newestFirstSelectedRows: newestFirst.length, + oldestFirstSelectedRows: oldestFirst.length, + newestFirstSelectedNewest, + oldestFirstSelectedNewest, + }, + }; +} + +async function runScenario(options: WorkerOptions, stateDir: string): Promise { + const timingsMs: number[] = []; + let invariant: Record = {}; + const rssStartBytes = process.memoryUsage().rss; + for (let index = 0; index < options.warmup + options.runs; index += 1) { + let sample: Sample; + if (options.scenario === "spawnPipelineInMemory") { + sample = await runSpawnSample(options.size, index, "memory", stateDir); + } else if (options.scenario === "spawnPipelineDurable") { + sample = await runSpawnSample(options.size, index, "durable", stateDir); + } else if (options.scenario === "admission") { + sample = await runAdmissionSample(options.size, index); + } else if (options.scenario === "recoverySweep") { + sample = await runSweepSample(options.size); + } else { + sample = await runDedupeSample(options.size); + } + invariant = sample.invariant; + if (index >= options.warmup) { + timingsMs.push(sample.durationMs); + } + } + return { + scenario: options.scenario, + size: options.size, + timingsMs, + memory: { + rssStartBytes, + rssEndBytes: process.memoryUsage().rss, + processMaxRssBytes: processMaxRssBytes(), + }, + invariant, + }; +} + +async function main(): Promise { + const options = parseOptions(process.argv.slice(2)); + const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-agent-concurrency-worker-")); + const previousStateDir = process.env.OPENCLAW_STATE_DIR; + const previousNodeEnv = process.env.NODE_ENV; + let result: WorkerResult | undefined; + let failure: unknown; + process.env.OPENCLAW_STATE_DIR = stateDir; + process.env.NODE_ENV = "test"; + try { + const { pinRuntimePaths } = await import("../src/config/paths.js"); + pinRuntimePaths(); + result = await runScenario(options, stateDir); + } catch (error) { + failure = error; + } finally { + try { + await resetRuntime(false); + } catch (error) { + failure ??= error; + } finally { + if (previousStateDir === undefined) { + delete process.env.OPENCLAW_STATE_DIR; + } else { + process.env.OPENCLAW_STATE_DIR = previousStateDir; + } + if (previousNodeEnv === undefined) { + delete process.env.NODE_ENV; + } else { + process.env.NODE_ENV = previousNodeEnv; + } + try { + fs.rmSync(stateDir, { recursive: true, force: true }); + } catch (error) { + failure ??= error; + } + } + } + if (failure) { + throw toError(failure); + } + if (!result) { + throw new Error("benchmark worker completed without a result"); + } + process.stdout.write(`${WORKER_RESULT_SENTINEL}${JSON.stringify(result)}\n`); +} + +if (import.meta.url === pathToFileURL(process.argv[1] ?? "").href) { + try { + await main(); + } catch (error) { + console.error(error instanceof Error ? error.stack : String(error)); + process.exitCode = 1; + } finally { + if (process.exitCode && process.exitCode !== 0) { + console.error(`[bench-agent-concurrency-worker] FAILED (exit ${process.exitCode})`); + } + } +} diff --git a/scripts/bench-agent-concurrency.ts b/scripts/bench-agent-concurrency.ts index bcdd2c29d836..517fea2032bd 100644 --- a/scripts/bench-agent-concurrency.ts +++ b/scripts/bench-agent-concurrency.ts @@ -1,12 +1,32 @@ +import { spawnSync } from "node:child_process"; import fs from "node:fs"; -import os from "node:os"; import path from "node:path"; -import { performance } from "node:perf_hooks"; import { pathToFileURL } from "node:url"; -import type { SubagentRunRecord } from "../src/agents/subagent-registry.types.js"; +import { isRecord } from "@openclaw/normalization-core/record-coerce"; -const DEFAULT_FANOUT = [1, 8, 32, 64], - DEFAULT_SWEEP_ROWS = [32, 128, 512]; +const DEFAULT_FANOUT = [1, 8, 32, 64]; +const DEFAULT_SWEEP_ROWS = [32, 128, 512]; +const WORKER_TIMEOUT_MS = 300_000; +export const WORKER_RESULT_SENTINEL = "[bench-agent-concurrency-result] "; + +export type WorkerScenario = + | "spawnPipelineInMemory" + | "spawnPipelineDurable" + | "admission" + | "recoverySweep" + | "duplicateSuppression"; + +export type WorkerResult = { + scenario: WorkerScenario; + size: number; + timingsMs: number[]; + memory: { + rssStartBytes: number; + rssEndBytes: number; + processMaxRssBytes: number; + }; + invariant: Record; +}; type Options = { runs: number; @@ -18,12 +38,90 @@ type Options = { help: boolean; }; -type TimingSummary = Record<"count" | "min" | "p50" | "p95" | "p99" | "max", number>; +type TimingSummary = { + count: number; + min: number; + p50: number; + max: number; + p95?: number; + p99?: number; +}; -type ScenarioResult = { - size: number; - timingsMs: TimingSummary; - invariant: Record; +const SCENARIO_SPECS: ReadonlyArray<{ + scenario: WorkerScenario; + sizes: "fanout" | "sweepRows"; +}> = [ + { scenario: "spawnPipelineInMemory", sizes: "fanout" }, + { scenario: "spawnPipelineDurable", sizes: "fanout" }, + { scenario: "admission", sizes: "fanout" }, + { scenario: "recoverySweep", sizes: "sweepRows" }, + { scenario: "duplicateSuppression", sizes: "sweepRows" }, +]; + +const REQUIRED_INVARIANT_FIELDS: Record = { + spawnPipelineInMemory: [ + "ok", + "registeredRuns", + "reservationsReleased", + "blockedWaits", + "settledRuns", + "settledTasks", + "outstandingWaits", + "durableSubagentRows", + "durableTaskRows", + "durableStateFile", + "postTeardownRegistryRows", + "postTeardownTaskRows", + "postTeardownDurableSubagentRows", + "postTeardownDurableTaskRows", + ], + spawnPipelineDurable: [ + "ok", + "registeredRuns", + "reservationsReleased", + "blockedWaits", + "settledRuns", + "settledTasks", + "outstandingWaits", + "durableSubagentRows", + "durableTaskRows", + "durableStateFile", + "postTeardownRegistryRows", + "postTeardownTaskRows", + "postTeardownDurableSubagentRows", + "postTeardownDurableTaskRows", + ], + admission: ["ok", "admissionCap", "overflowRejected", "released"], + recoverySweep: [ + "ok", + "seededRows", + "removedRows", + "retainedCurrent", + "sessionEffects", + "recoveryProjections", + "lostContextCompletions", + ], + duplicateSuppression: [ + "ok", + "inputRowsPerOrdering", + "newestFirstSelectedRows", + "oldestFirstSelectedRows", + "newestFirstSelectedNewest", + "oldestFirstSelectedNewest", + ], +}; + +type WorkerProcessResult = { + status: number | null; + stdout: string; + stderr: string; + error?: Error & { code?: string }; +}; + +type BenchmarkRuntime = { + runWorker?: typeof runWorker; + writeProgress?: (line: string) => void; + now?: () => number; }; function usage(): string { @@ -124,390 +222,252 @@ function summarizeTimings(values: number[]): TimingSummary { throw new Error("cannot summarize an empty timing set"); } const sorted = values.toSorted((left, right) => left - right); - return { + const summary: TimingSummary = { count: sorted.length, min: sorted[0] ?? 0, p50: percentile(sorted, 0.5), - p95: percentile(sorted, 0.95), - p99: percentile(sorted, 0.99), max: sorted.at(-1) ?? 0, }; + if (sorted.length >= 20) { + summary.p95 = percentile(sorted, 0.95); + summary.p99 = percentile(sorted, 0.99); + } + return summary; } -async function sampleScenario( +function expectedWorkerKeys(options: Options): string[] { + return SCENARIO_SPECS.flatMap(({ scenario, sizes }) => + options[sizes].map((size) => `${scenario}:${size}`), + ); +} + +function aggregateWorkerResults( options: Options, - sample: () => Promise, - validate: (result: T) => Record, -): Promise<{ timingsMs: TimingSummary; invariant: Record }> { - const timings: number[] = []; - let invariant: Record = {}; - for (let index = 0; index < options.warmup + options.runs; index += 1) { - const result = await sample(); - invariant = validate(result); - if (index >= options.warmup) { - timings.push(result.durationMs); - } + workers: WorkerResult[], + parentMemory = { + rssStartBytes: process.memoryUsage().rss, + rssEndBytes: process.memoryUsage().rss, + }, +) { + const expected = expectedWorkerKeys(options); + const byKey = new Map(workers.map((worker) => [`${worker.scenario}:${worker.size}`, worker])); + if (byKey.size !== workers.length) { + throw new Error("worker results contain duplicate scenario/size pairs"); + } + const missing = expected.filter((key) => !byKey.has(key)); + const unexpected = [...byKey.keys()].filter((key) => !expected.includes(key)); + if (missing.length > 0 || unexpected.length > 0) { + throw new Error( + `worker result mismatch: missing=${missing.join(",") || "none"} unexpected=${unexpected.join(",") || "none"}`, + ); } - return { timingsMs: summarizeTimings(timings), invariant }; -} -async function runSpawnPipelineSample(fanout: number, serial: number) { - const [{ runSpawnPipeline }, registry, helpers] = await Promise.all([ - import("../src/agents/spawn-pipeline.js"), - import("../src/agents/subagent-registry-memory.js"), - import("../src/agents/subagent-registry.test-helpers.js"), - ]); - helpers.resetSubagentRegistryForTests({ persist: false }); - helpers.testing.setDepsForTest({ - getRuntimeConfig: () => ({}), - onAgentEvent: () => () => {}, - persistSubagentRunsToDisk: () => {}, - persistSubagentRunsToDiskOrThrow: () => {}, - }); - let releases = 0; - const pipelineParams = Array.from({ length: fanout }, (_, index) => { - const runId = `bench-pipeline-${serial}-${index}`; - let released = false; - return { - adapter: { - initialize: async () => ({ index }), - dispatchTurn: async () => ({ runId }), - cleanupOnFailure: async () => {}, - }, - admissionReservation: { - release: () => { - if (!released) { - released = true; - releases += 1; - } - }, - }, - buildRegistration: () => ({ - runId, - childSessionKey: `agent:bench:subagent:${serial}:${index}`, - requesterSessionKey: "agent:bench:main", - requesterDisplayKey: "bench", - task: `benchmark child ${index}`, - cleanup: "keep" as const, - expectsCompletionMessage: false, - }), - progressSessionKey: "agent:bench:main", - }; - }); - const startedAt = performance.now(); - const results = await Promise.all(pipelineParams.map((params) => runSpawnPipeline(params))); - const durationMs = performance.now() - startedAt; - const registered = results.filter((result) => result.ok).map((result) => result.runId); - const uniqueRegistered = new Set(registered).size; - const registrySize = registry.subagentRuns.size; - helpers.resetSubagentRegistryForTests({ persist: false }); - helpers.testing.setDepsForTest(); - return { durationMs, expected: fanout, uniqueRegistered, registrySize, releases }; -} - -function validateSpawnPipeline(result: Awaited>) { - const ok = - result.uniqueRegistered === result.expected && - result.registrySize === result.expected && - result.releases === result.expected; - if (!ok) { - throw new Error(`spawn pipeline invariant failed: ${JSON.stringify(result)}`); - } - return { ok, registeredRuns: result.registrySize, reservationsReleased: result.releases }; -} - -async function runAdmissionSample(fanout: number, serial: number) { - const { reserveChildAdmissionSlot } = await import("../src/agents/child-admission.js"); - const controllerSessionKey = `agent:bench:admission:${serial}`; - const reservations: Array<{ release: () => void }> = []; - const reserve = () => - reserveChildAdmissionSlot({ - controllerSessionKey, - resolveAdmission: (pendingChildren) => - pendingChildren < fanout - ? { ok: true as const } - : { ok: false as const, governingCap: "benchmark" }, - }); - const startedAt = performance.now(); - for (let index = 0; index < fanout; index += 1) { - const reservation = reserve(); - if (!reservation.ok) { - throw new Error(`admission rejected slot ${index + 1}/${fanout}`); - } - reservations.push(reservation); - } - const overflow = reserve(); - for (const reservation of reservations) { - reservation.release(); - reservation.release(); - } - const replacement = reserve(); - if (replacement.ok) { - replacement.release(); - } - return { - durationMs: performance.now() - startedAt, - admitted: reservations.length, - overflowRejected: !overflow.ok, - replacement: replacement.ok, - }; -} - -function validateAdmission(result: Awaited>) { - const ok = result.overflowRejected && result.replacement; - if (!ok) { - throw new Error(`admission invariant failed: ${JSON.stringify(result)}`); - } - return { ok, admissionCap: result.admitted, overflowRejected: true, released: true }; -} - -function recoveryRow(child: number, generation: number, now: number): SubagentRunRecord { - const current = generation === 3; - return { - runId: `bench-sweep-${child}-${generation}`, - childSessionKey: `agent:bench:subagent:sweep-${child}`, - requesterSessionKey: "agent:bench:main", - requesterDisplayKey: "bench", - task: `sweep child ${child}`, - cleanup: current ? "keep" : "delete", - generation, - createdAt: now - generation, - archiveAtMs: current ? undefined : now - 1, - terminalOwner: current ? "interrupted-recovery" : undefined, - endedReason: current ? "subagent-error" : undefined, - execution: current - ? { - status: "terminal", - startedAt: now - 2_000, - endedAt: now - 1_000, - outcome: { status: "error", error: "interrupted recovery replay" }, - suppressSessionEffects: true, + const scenarios = Object.fromEntries( + SCENARIO_SPECS.map(({ scenario, sizes }) => [ + scenario, + options[sizes].map((size) => { + const worker = byKey.get(`${scenario}:${size}`); + if (!worker) { + throw new Error(`missing worker result for ${scenario}:${size}`); } - : { - status: "terminal", - startedAt: now - 2_000, - endedAt: now - 1_000, - outcome: { status: "error", error: "retired recovery generation" }, - suppressSessionEffects: true, - }, + return { + size, + timingsMs: summarizeTimings(worker.timingsMs), + memory: worker.memory, + invariant: worker.invariant, + }; + }), + ]), + ) as Record< + WorkerScenario, + Array<{ + size: number; + timingsMs: TimingSummary; + memory: WorkerResult["memory"]; + invariant: WorkerResult["invariant"]; + }> + >; + + const checks = { + spawnPipelineInMemory: scenarios.spawnPipelineInMemory.every( + (entry) => entry.invariant.ok === true, + ), + spawnPipelineDurable: scenarios.spawnPipelineDurable.every( + (entry) => entry.invariant.ok === true, + ), + admissionCapOverflowRelease: scenarios.admission.every((entry) => entry.invariant.ok === true), + sweepRecoveryRowsWithoutSessionEffects: scenarios.recoverySweep.every( + (entry) => entry.invariant.ok === true, + ), + dedupeNewestPerChild: scenarios.duplicateSuppression.every( + (entry) => entry.invariant.ok === true, + ), + }; + const failures = Object.entries(checks) + .filter(([, ok]) => !ok) + .map(([name]) => name); + + return { + schemaVersion: 2, + generatedAt: new Date().toISOString(), + runtime: { node: process.version, platform: process.platform, arch: process.arch }, + options: { + runs: options.runs, + warmup: options.warmup, + fanout: options.fanout, + sweepRows: options.sweepRows, + }, + memory: { + ...parentMemory, + workerProcessMaxRssBytes: Math.max( + ...workers.map((worker) => worker.memory.processMaxRssBytes), + ), + }, + scenarios, + invariants: { ok: failures.length === 0, failures, ...checks }, }; } -async function runSweepSample(childCount: number) { - const { createSubagentRegistrySweeper } = - await import("../src/agents/subagent-registry-sweeper.js"); - const now = Date.now(); - const runs = new Map(); - for (let child = 0; child < childCount; child += 1) { - for (const generation of [3, 2, 1]) { - const entry = recoveryRow(child, generation, now); - runs.set(entry.runId, entry); +function assertFiniteNonNegative(value: unknown, field: string): asserts value is number { + if (typeof value !== "number" || !Number.isFinite(value) || value < 0) { + throw new Error(`worker result field ${field} must be a finite nonnegative number`); + } +} + +function validateWorkerResult( + value: unknown, + expected: { scenario: WorkerScenario; size: number; runs: number }, +): WorkerResult { + if (!isRecord(value)) { + throw new Error("worker result must be an object"); + } + if (value.scenario !== expected.scenario || value.size !== expected.size) { + throw new Error(`worker ${expected.scenario}:${expected.size} returned mismatched identity`); + } + if (!Array.isArray(value.timingsMs) || value.timingsMs.length !== expected.runs) { + throw new Error( + `worker ${expected.scenario}:${expected.size} returned ${Array.isArray(value.timingsMs) ? value.timingsMs.length : "invalid"} samples; expected ${expected.runs}`, + ); + } + value.timingsMs.forEach((timing, index) => + assertFiniteNonNegative(timing, `timingsMs[${index}]`), + ); + if (!isRecord(value.memory)) { + throw new Error("worker result memory must be an object"); + } + assertFiniteNonNegative(value.memory.rssStartBytes, "memory.rssStartBytes"); + assertFiniteNonNegative(value.memory.rssEndBytes, "memory.rssEndBytes"); + assertFiniteNonNegative(value.memory.processMaxRssBytes, "memory.processMaxRssBytes"); + if (!isRecord(value.invariant)) { + throw new Error("worker result invariant must be an object"); + } + for (const field of REQUIRED_INVARIANT_FIELDS[expected.scenario]) { + const invariantValue = value.invariant[field]; + if (typeof invariantValue !== "number" && typeof invariantValue !== "boolean") { + throw new Error(`worker result invariant.${field} is missing or invalid`); } } - let sessionEffects = 0; - let recoveryProjections = 0; - let lostContextCompletions = 0; - const sweeper = createSubagentRegistrySweeper({ - runs, - resumedRuns: new Set(), - persist: () => {}, - clearPendingLifecycleError: () => {}, - clearPendingLifecycleTimeout: () => {}, - sweepPendingLifecycle: () => {}, - completeSubagentRunWithRecovery: async () => { - lostContextCompletions += 1; + if (value.invariant.ok !== true) { + throw new Error(`worker ${expected.scenario}:${expected.size} reported a failed invariant`); + } + return value as WorkerResult; +} + +function parseWorkerProcessResult( + result: WorkerProcessResult, + expected: { scenario: WorkerScenario; size: number; runs: number }, +): WorkerResult { + if (result.error) { + const detail = + "code" in result.error && result.error.code === "ETIMEDOUT" + ? `timed out after ${WORKER_TIMEOUT_MS}ms` + : result.error.message; + throw new Error(`worker ${expected.scenario}:${expected.size} failed: ${detail}`); + } + if (result.status !== 0) { + throw new Error( + `worker ${expected.scenario}:${expected.size} failed (${result.status ?? "signal"}): ${result.stderr.trim() || result.stdout.trim()}`, + ); + } + const payloads = result.stdout + .split(/\r?\n/u) + .filter((line) => line.startsWith(WORKER_RESULT_SENTINEL)); + if (payloads.length !== 1) { + throw new Error( + `worker ${expected.scenario}:${expected.size} returned ${payloads.length} result payloads`, + ); + } + let parsed: unknown; + try { + parsed = JSON.parse(payloads[0]!.slice(WORKER_RESULT_SENTINEL.length)); + } catch { + throw new Error(`worker ${expected.scenario}:${expected.size} returned invalid JSON`); + } + return validateWorkerResult(parsed, expected); +} + +function runWorker(options: Options, scenario: WorkerScenario, size: number): WorkerResult { + const result = spawnSync( + process.execPath, + [ + "--import", + "tsx", + "scripts/bench-agent-concurrency-worker.ts", + "--scenario", + scenario, + "--size", + String(size), + "--runs", + String(options.runs), + "--warmup", + String(options.warmup), + ], + { + cwd: process.cwd(), + encoding: "utf8", + env: { ...process.env, NODE_NO_WARNINGS: "1" }, + timeout: WORKER_TIMEOUT_MS, + killSignal: "SIGTERM", + maxBuffer: 16 * 1024 * 1024, }, - getGatewayRecoveryRuntime: () => undefined, - abandonSubagentRestartRecoveryLaunch: () => true, - clearAcceptedSubagentRestartRecovery: () => true, - resumeSettledSubagentRestartRecovery: () => true, - replaceSubagentRunAfterSteer: () => true, - markSubagentRestartRecoveryLaunchAttempted: () => undefined, - markSubagentRestartRecoveryLaunchAccepted: () => undefined, - markSubagentRestartRecoveryLaunchConsumed: () => undefined, - reserveSubagentRestartRecoveryLaunch: () => undefined, - resetSubagentRestartRecoveryLaunchAttempt: () => true, - finalizeInterruptedSubagentRun: async ({ runId, expectedEntry }) => { - if (runs.get(runId) !== expectedEntry || expectedEntry?.generation !== 3) { - throw new Error(`unexpected recovery projection owner: ${runId}`); - } - recoveryProjections += 1; - return 1; - }, - resumeRequesterSettleWake: () => {}, - startSubagentAnnounceCleanupFlow: () => true, - completeCleanupBookkeeping: () => {}, - shouldEmitEndedHookForRun: () => false, - emitSubagentEndedHookForRun: async () => {}, - callGateway: (async () => { - sessionEffects += 1; - return {} as T; - }) as typeof import("../src/gateway/call.js").callGateway, - cleanupCollectorLaunchResources: async () => true, - runContextEngineSubagentEnded: async () => { - sessionEffects += 1; - }, - notifyContextEngineSubagentEnded: async () => { - sessionEffects += 1; - }, - retireSupersededRun: async () => {}, - getRunsForChildSession: (childSessionKey) => - [...runs.values()].filter((entry) => entry.childSessionKey === childSessionKey), - getRunsForCollectorGroup: () => [], - warn: () => {}, + ); + return parseWorkerProcessResult(result, { scenario, size, runs: options.runs }); +} + +function benchmark(options: Options, runtime: BenchmarkRuntime = {}) { + const rssStartBytes = process.memoryUsage().rss; + const jobs = SCENARIO_SPECS.flatMap(({ scenario, sizes }) => + options[sizes].map((size) => ({ scenario, size })), + ); + const run = runtime.runWorker ?? runWorker; + const writeProgress = + runtime.writeProgress ?? ((line: string) => process.stderr.write(`${line}\n`)); + const now = runtime.now ?? Date.now; + const workers = jobs.map(({ scenario, size }, index) => { + const ordinal = index + 1; + writeProgress( + `[bench-agent-concurrency] worker ${ordinal}/${jobs.length} start scenario=${scenario} size=${size}`, + ); + const startedAt = now(); + try { + const worker = run(options, scenario, size); + const elapsedMs = Math.max(0, now() - startedAt); + writeProgress( + `[bench-agent-concurrency] worker ${ordinal}/${jobs.length} complete scenario=${scenario} size=${size} elapsed=${(elapsedMs / 1_000).toFixed(3)}s`, + ); + return worker; + } catch (error) { + const elapsedMs = Math.max(0, now() - startedAt); + writeProgress( + `[bench-agent-concurrency] worker ${ordinal}/${jobs.length} failed scenario=${scenario} size=${size} elapsed=${(elapsedMs / 1_000).toFixed(3)}s`, + ); + throw error; + } + }); + return aggregateWorkerResults(options, workers, { + rssStartBytes, + rssEndBytes: process.memoryUsage().rss, }); - const startedAt = performance.now(); - let durationMs: number; - try { - await sweeper.sweepOnce(); - durationMs = performance.now() - startedAt; - } finally { - sweeper.reset(); - } - const retainedCurrent = [...runs.values()].filter((entry) => entry.generation === 3).length; - return { - durationMs, - seededRows: childCount * 3, - removedRows: childCount * 3 - runs.size, - retainedCurrent, - sessionEffects, - recoveryProjections, - lostContextCompletions, - }; -} - -function validateSweep(result: Awaited>) { - const expectedChildren = result.seededRows / 3; - const ok = - result.removedRows === expectedChildren * 2 && - result.retainedCurrent === expectedChildren && - result.sessionEffects === 0 && - result.recoveryProjections === expectedChildren && - result.lostContextCompletions === 0; - if (!ok) { - throw new Error(`registry sweep invariant failed: ${JSON.stringify(result)}`); - } - return { ok, ...result }; -} - -async function runDedupeSample(childCount: number) { - const { dedupeLatestChildCompletionRows } = - await import("../src/agents/subagent-announce-output.js"); - const rows = Array.from({ length: childCount }, (_, child) => - [3, 2, 1].map((generation) => ({ - runId: `bench-dedupe-${child}-${generation}`, - childSessionKey: `agent:bench:subagent:dedupe-${child}`, - task: `dedupe child ${child}`, - generation, - createdAt: generation, - execution: { status: "terminal" as const, endedAt: generation }, - })), - ).flat(); - const startedAt = performance.now(); - const deduped = dedupeLatestChildCompletionRows(rows); - return { - durationMs: performance.now() - startedAt, - inputRows: rows.length, - selectedRows: deduped.length, - newestSelected: deduped.every((row) => row.generation === 3), - }; -} - -function validateDedupe(result: Awaited>) { - const ok = result.selectedRows === result.inputRows / 3 && result.newestSelected; - if (!ok) { - throw new Error(`completion dedupe invariant failed: ${JSON.stringify(result)}`); - } - return { ok, ...result }; -} - -async function benchmark(options: Options) { - const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-agent-concurrency-")); - const previousStateDir = process.env.OPENCLAW_STATE_DIR; - const previousNodeEnv = process.env.NODE_ENV; - process.env.OPENCLAW_STATE_DIR = stateDir; - process.env.NODE_ENV = "test"; - let serial = 0; - let peakRss = process.memoryUsage().rss; - const beforeRss = peakRss; - const runSized = async ( - sizes: number[], - sample: (size: number, serial: number) => Promise, - validate: (result: T) => Record, - ): Promise => { - const results: ScenarioResult[] = []; - for (const size of sizes) { - const measured = await sampleScenario(options, () => sample(size, serial++), validate); - peakRss = Math.max(peakRss, process.memoryUsage().rss); - results.push({ size, ...measured }); - } - return results; - }; - try { - const scenarios = { - spawnPipeline: await runSized(options.fanout, runSpawnPipelineSample, validateSpawnPipeline), - admission: await runSized(options.fanout, runAdmissionSample, validateAdmission), - recoverySweep: await runSized( - options.sweepRows, - (size) => runSweepSample(size), - validateSweep, - ), - duplicateSuppression: await runSized( - options.sweepRows, - (size) => runDedupeSample(size), - validateDedupe, - ), - }; - const afterRss = process.memoryUsage().rss; - const checks = { - admissionCapOverflowRelease: scenarios.admission.every((entry) => entry.invariant.ok), - uniqueRegisteredRunsAndReleasedReservations: scenarios.spawnPipeline.every( - (entry) => entry.invariant.ok, - ), - sweepRecoveryRowsWithoutSessionEffects: scenarios.recoverySweep.every( - (entry) => entry.invariant.ok, - ), - dedupeNewestPerChild: scenarios.duplicateSuppression.every((entry) => entry.invariant.ok), - }; - const failures = Object.entries(checks) - .filter(([, ok]) => !ok) - .map(([name]) => name); - return { - schemaVersion: 1, - generatedAt: new Date().toISOString(), - runtime: { node: process.version, platform: process.platform, arch: process.arch }, - options: { - runs: options.runs, - warmup: options.warmup, - fanout: options.fanout, - sweepRows: options.sweepRows, - }, - memory: { - rssStartBytes: beforeRss, - rssPeakBytes: Math.max(peakRss, afterRss), - rssEndBytes: afterRss, - rssDeltaBytes: afterRss - beforeRss, - }, - scenarios, - invariants: { - ok: failures.length === 0, - failures, - ...checks, - }, - }; - } finally { - if (previousStateDir === undefined) { - delete process.env.OPENCLAW_STATE_DIR; - } else { - process.env.OPENCLAW_STATE_DIR = previousStateDir; - } - if (previousNodeEnv === undefined) { - delete process.env.NODE_ENV; - } else { - process.env.NODE_ENV = previousNodeEnv; - } - fs.rmSync(stateDir, { recursive: true, force: true }); - } } async function main(argv = process.argv.slice(2)): Promise { @@ -516,7 +476,7 @@ async function main(argv = process.argv.slice(2)): Promise { process.stdout.write(usage()); return; } - const report = await benchmark(options); + const report = benchmark(options); const json = `${JSON.stringify(report, null, 2)}\n`; if (options.output) { fs.mkdirSync(path.dirname(path.resolve(options.output)), { recursive: true }); @@ -528,17 +488,25 @@ async function main(argv = process.argv.slice(2)): Promise { } for (const [name, scenarios] of Object.entries(report.scenarios)) { for (const scenario of scenarios) { + const tail = + scenario.timingsMs.p95 === undefined + ? "" + : ` p95=${scenario.timingsMs.p95.toFixed(3)}ms p99=${scenario.timingsMs.p99?.toFixed(3)}ms`; console.log( - `${name} size=${scenario.size} p50=${scenario.timingsMs.p50.toFixed(3)}ms p95=${scenario.timingsMs.p95.toFixed(3)}ms`, + `${name} size=${scenario.size} p50=${scenario.timingsMs.p50.toFixed(3)}ms max=${scenario.timingsMs.max.toFixed(3)}ms${tail}`, ); } } - console.log(`peak RSS ${(report.memory.rssPeakBytes / 1024 / 1024).toFixed(1)} MiB`); + console.log( + `max worker RSS ${(report.memory.workerProcessMaxRssBytes / 1024 / 1024).toFixed(1)} MiB`, + ); } export const testing = { + aggregateWorkerResults, benchmark, parseOptions, + parseWorkerProcessResult, summarizeTimings, }; diff --git a/test/scripts/bench-agent-concurrency.test.ts b/test/scripts/bench-agent-concurrency.test.ts index 5d4e37e711db..595af27deb32 100644 --- a/test/scripts/bench-agent-concurrency.test.ts +++ b/test/scripts/bench-agent-concurrency.test.ts @@ -1,6 +1,67 @@ import { spawnSync } from "node:child_process"; import { describe, expect, it } from "vitest"; -import { testing } from "../../scripts/bench-agent-concurrency.ts"; +import { + testing, + WORKER_RESULT_SENTINEL, + type WorkerResult, + type WorkerScenario, +} from "../../scripts/bench-agent-concurrency.ts"; + +function workerResult(scenario: WorkerScenario, size: number, timingsMs = [1, 2, 3]): WorkerResult { + const invariant: Record = + scenario === "spawnPipelineInMemory" || scenario === "spawnPipelineDurable" + ? { + ok: true, + registeredRuns: size, + reservationsReleased: size, + blockedWaits: size, + settledRuns: size, + settledTasks: size, + outstandingWaits: 0, + durableSubagentRows: scenario === "spawnPipelineDurable" ? size : 0, + durableTaskRows: scenario === "spawnPipelineDurable" ? size : 0, + durableStateFile: scenario === "spawnPipelineDurable", + postTeardownRegistryRows: 0, + postTeardownTaskRows: 0, + postTeardownDurableSubagentRows: 0, + postTeardownDurableTaskRows: 0, + } + : scenario === "admission" + ? { ok: true, admissionCap: size, overflowRejected: true, released: true } + : scenario === "recoverySweep" + ? { + ok: true, + seededRows: size * 3, + removedRows: size * 2, + retainedCurrent: size, + sessionEffects: 0, + recoveryProjections: size, + lostContextCompletions: 0, + } + : { + ok: true, + inputRowsPerOrdering: size * 3, + newestFirstSelectedRows: size, + oldestFirstSelectedRows: size, + newestFirstSelectedNewest: true, + oldestFirstSelectedNewest: true, + }; + return { + scenario, + size, + timingsMs, + memory: { + rssStartBytes: 100, + rssEndBytes: 120, + processMaxRssBytes: 150, + }, + invariant, + }; +} + +function workerStdout(result: WorkerResult): string { + return `${WORKER_RESULT_SENTINEL}${JSON.stringify(result)}\n`; +} describe("agent concurrency benchmark", () => { it("parses bounded options and rejects ambiguous arguments", () => { @@ -36,60 +97,189 @@ describe("agent concurrency benchmark", () => { expect(() => testing.parseOptions(["--wat"])).toThrow("Unknown argument: --wat"); }); - it("summarizes min and nearest-rank percentiles", () => { + it("emits tail percentiles only when the sample supports them", () => { expect(testing.summarizeTimings([100, 1, 4, 2, 3])).toEqual({ count: 5, min: 1, p50: 3, - p95: 100, - p99: 100, max: 100, }); + const summary = testing.summarizeTimings(Array.from({ length: 20 }, (_, index) => index + 1)); + expect(summary).toMatchObject({ count: 20, p50: 10, p95: 19, p99: 20, max: 20 }); }); - it("emits schema version 1, RSS, scenarios, and invariants in a tiny real smoke", async () => { - const previousNodeEnv = process.env.NODE_ENV; - const report = await testing.benchmark( - testing.parseOptions(["--runs", "1", "--warmup", "0", "--fanout", "2", "--sweep-rows", "2"]), + it("aggregates synthetic worker results into schema version 2", () => { + const options = testing.parseOptions([ + "--runs", + "3", + "--warmup", + "1", + "--fanout", + "2", + "--sweep-rows", + "4", + ]); + const report = testing.aggregateWorkerResults( + options, + [ + workerResult("spawnPipelineInMemory", 2), + workerResult("spawnPipelineDurable", 2), + workerResult("admission", 2), + workerResult("recoverySweep", 4), + workerResult("duplicateSuppression", 4), + ], + { rssStartBytes: 10, rssEndBytes: 20 }, ); expect(report).toMatchObject({ - schemaVersion: 1, - runtime: { node: process.version }, - options: { runs: 1, warmup: 0, fanout: [2], sweepRows: [2] }, + schemaVersion: 2, + options: { runs: 3, warmup: 1, fanout: [2], sweepRows: [4] }, memory: { - rssStartBytes: expect.any(Number), - rssPeakBytes: expect.any(Number), - rssEndBytes: expect.any(Number), - rssDeltaBytes: expect.any(Number), + rssStartBytes: 10, + rssEndBytes: 20, + workerProcessMaxRssBytes: 150, }, invariants: { ok: true, failures: [], + spawnPipelineInMemory: true, + spawnPipelineDurable: true, admissionCapOverflowRelease: true, - uniqueRegisteredRunsAndReleasedReservations: true, sweepRecoveryRowsWithoutSessionEffects: true, dedupeNewestPerChild: true, }, }); - expect(process.env.NODE_ENV).toBe(previousNodeEnv); + expect(report.scenarios.spawnPipelineDurable[0]?.timingsMs).toEqual({ + count: 3, + min: 1, + p50: 2, + max: 3, + }); expect(report.generatedAt).toEqual(expect.any(String)); - expect(report.scenarios.spawnPipeline[0]?.timingsMs.count).toBe(1); - expect(report.scenarios.recoverySweep[0]?.invariant).toMatchObject({ - seededRows: 6, - removedRows: 4, - retainedCurrent: 2, - sessionEffects: 0, - recoveryProjections: 2, - lostContextCompletions: 0, - }); - expect(report.scenarios.duplicateSuppression[0]?.invariant).toMatchObject({ - selectedRows: 2, - newestSelected: true, - }); }); - it("supports native Node TypeScript help and ends failures with the marker", () => { + it("reports deterministic parent progress around every worker", () => { + const options = testing.parseOptions([ + "--runs", + "3", + "--warmup", + "0", + "--fanout", + "2", + "--sweep-rows", + "4", + ]); + const progress: string[] = []; + let now = 0; + const report = testing.benchmark(options, { + runWorker: (_options, scenario, size) => workerResult(scenario, size), + writeProgress: (line) => progress.push(line), + now: () => { + const value = now; + now += 250; + return value; + }, + }); + + expect(report.invariants.ok).toBe(true); + expect(progress).toHaveLength(10); + expect(progress[0]).toBe( + "[bench-agent-concurrency] worker 1/5 start scenario=spawnPipelineInMemory size=2", + ); + expect(progress[1]).toBe( + "[bench-agent-concurrency] worker 1/5 complete scenario=spawnPipelineInMemory size=2 elapsed=0.250s", + ); + expect(progress.at(-2)).toBe( + "[bench-agent-concurrency] worker 5/5 start scenario=duplicateSuppression size=4", + ); + expect(progress.at(-1)).toBe( + "[bench-agent-concurrency] worker 5/5 complete scenario=duplicateSuppression size=4 elapsed=0.250s", + ); + }); + + it("rejects incomplete synthetic worker sets", () => { + const options = testing.parseOptions(["--fanout", "1", "--sweep-rows", "1"]); + expect(() => + testing.aggregateWorkerResults(options, [workerResult("spawnPipelineInMemory", 1)]), + ).toThrow("worker result mismatch"); + }); + + it("rejects malformed, duplicate, mismatched, partial, and timed-out worker results", () => { + const expected = { scenario: "admission" as const, size: 2, runs: 3 }; + const valid = workerResult("admission", 2); + expect( + testing.parseWorkerProcessResult( + { status: 0, stdout: workerStdout(valid), stderr: "" }, + expected, + ), + ).toEqual(valid); + + expect(() => + testing.parseWorkerProcessResult( + { status: 0, stdout: `${WORKER_RESULT_SENTINEL}{\n`, stderr: "" }, + expected, + ), + ).toThrow("returned invalid JSON"); + expect(() => + testing.parseWorkerProcessResult( + { status: 0, stdout: `${workerStdout(valid)}${workerStdout(valid)}`, stderr: "" }, + expected, + ), + ).toThrow("returned 2 result payloads"); + expect(() => + testing.parseWorkerProcessResult( + { status: 0, stdout: workerStdout(workerResult("admission", 3)), stderr: "" }, + expected, + ), + ).toThrow("mismatched identity"); + expect(() => + testing.parseWorkerProcessResult( + { + status: 0, + stdout: `${WORKER_RESULT_SENTINEL}${JSON.stringify({ ...valid, memory: {} })}\n`, + stderr: "", + }, + expected, + ), + ).toThrow("memory.rssStartBytes"); + expect(() => + testing.parseWorkerProcessResult( + { + status: 0, + stdout: workerStdout({ ...valid, timingsMs: [1, 2] }), + stderr: "", + }, + expected, + ), + ).toThrow("returned 2 samples; expected 3"); + const missingInvariant = structuredClone(valid); + delete missingInvariant.invariant.admissionCap; + expect(() => + testing.parseWorkerProcessResult( + { status: 0, stdout: workerStdout(missingInvariant), stderr: "" }, + expected, + ), + ).toThrow("invariant.admissionCap is missing or invalid"); + expect(() => + testing.parseWorkerProcessResult( + { status: 7, stdout: "", stderr: "worker exploded" }, + expected, + ), + ).toThrow("failed (7): worker exploded"); + expect(() => + testing.parseWorkerProcessResult( + { + status: null, + stdout: "", + stderr: "", + error: Object.assign(new Error("spawnSync timed out"), { code: "ETIMEDOUT" }), + }, + expected, + ), + ).toThrow("timed out after 300000ms"); + }); + + it("supports help and ends failures with the marker", () => { const help = spawnSync( process.execPath, ["--import", "tsx", "scripts/bench-agent-concurrency.ts", "--help"],