mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-12 21:53:00 -06:00
Peter Steinberger
b080dd1e76
* refactor: consolidate coercion contracts Centralize exact string, record, numeric, date, Boolean, argument, and structured-error coercions while preserving call-site semantics. Migrate canonical-name collisions and deprecated internal SDK bypasses, deleting 55 net production/tooling lines. Expand declaration ownership enforcement to 101 allowed helpers and add a narrow export-completeness audit. * fix: preserve standalone script coercions Keep copied Control UI tooling self-contained and retain the trusted release harness module-relative source seam when the harness runs against an old target cwd.
833 lines
29 KiB
TypeScript
833 lines
29 KiB
TypeScript
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 { toErrorObject } from "@openclaw/normalization-core/error-coercion";
|
|
import type { SubagentRunRecord } from "../src/agents/subagents/registry/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";
|
|
import { classifyBoundedUnsignedDecimal } from "./lib/arg-utils.mts";
|
|
|
|
type WorkerOptions = {
|
|
scenario: WorkerScenario;
|
|
size: number;
|
|
runs: number;
|
|
warmup: number;
|
|
};
|
|
|
|
type Sample = {
|
|
durationMs: number;
|
|
invariant: Record<string, number | boolean>;
|
|
};
|
|
|
|
const SCENARIOS = new Set<WorkerScenario>([
|
|
"spawnPipelineInMemory",
|
|
"spawnPipelineDurable",
|
|
"admission",
|
|
"recoverySweep",
|
|
"duplicateSuppression",
|
|
]);
|
|
|
|
function parseInteger(raw: string | undefined, flag: string, min: number, max: number): number {
|
|
const result = classifyBoundedUnsignedDecimal(raw, min, max);
|
|
if (result.kind === "syntax") {
|
|
throw new Error(`${flag} must be an integer`);
|
|
}
|
|
if (result.kind !== "value") {
|
|
throw new Error(`${flag} must be between ${min} and ${max}`);
|
|
}
|
|
return result.value;
|
|
}
|
|
|
|
function parseOptions(argv: string[]): WorkerOptions {
|
|
const values = new Map<string, string>();
|
|
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));
|
|
}
|
|
|
|
async function waitForCondition(check: () => boolean): Promise<boolean> {
|
|
const deadline = Date.now() + 30_000;
|
|
while (Date.now() < deadline) {
|
|
if (check()) {
|
|
return true;
|
|
}
|
|
await new Promise<void>((resolve) => {
|
|
setImmediate(resolve);
|
|
});
|
|
}
|
|
return check();
|
|
}
|
|
|
|
async function drainSpawnSampleRootWork(
|
|
waitForRootWork?: (timeoutMs: number) => Promise<{ drained: boolean; active: number }>,
|
|
): Promise<void> {
|
|
const wait =
|
|
waitForRootWork ??
|
|
(await import("../src/process/gateway-work-admission.js")).waitForActiveGatewayRootWork;
|
|
const result = await wait(30_000);
|
|
if (!result.drained || result.active !== 0) {
|
|
throw new Error(`spawn sample left ${result.active} active gateway root work items`);
|
|
}
|
|
}
|
|
|
|
async function resetRuntime(persist: boolean): Promise<void> {
|
|
const [subagents, tasks, stateDb, agentDb] = await Promise.all([
|
|
import("../src/agents/subagents/registry/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<string, (value: unknown) => void>();
|
|
let outstanding = 0;
|
|
let releasesStarted = false;
|
|
const callGateway = (async <T>(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<T>((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<void> {
|
|
const [subagents, registry, taskStore, flowStore] = await Promise.all([
|
|
import("../src/agents/subagents/registry/subagent-registry.test-helpers.js"),
|
|
import("../src/agents/subagents/registry/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 () => "retryable" as const,
|
|
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<OpenClawStateKyselyDatabase, "subagent_runs" | "task_runs">;
|
|
|
|
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<BenchmarkStateDatabase>(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<string | null>, 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<Sample> {
|
|
const [pipeline, registry] = await Promise.all([
|
|
import("../src/agents/spawn-pipeline.js"),
|
|
import("../src/agents/subagents/registry/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 })}`,
|
|
);
|
|
}
|
|
// Terminal rows can settle before detached cleanup and requester-wake roots.
|
|
// Drain before reset so leaked work stays visible and cannot reach the next sample.
|
|
await drainSpawnSampleRootWork();
|
|
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,
|
|
postTeardownActiveRootWork: 0,
|
|
},
|
|
};
|
|
} catch (error) {
|
|
failure = error;
|
|
} finally {
|
|
try {
|
|
barrier.releaseAll();
|
|
} catch (error) {
|
|
failure ??= error;
|
|
}
|
|
try {
|
|
await resetRuntime(mode === "durable");
|
|
} catch (error) {
|
|
failure ??= error;
|
|
}
|
|
}
|
|
if (failure) {
|
|
throw toErrorObject(failure, "Agent concurrency benchmark failed");
|
|
}
|
|
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<Sample> {
|
|
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<Sample> {
|
|
const [
|
|
{ getSubagentRunsForChildSession, subagentRuns: runs },
|
|
{ createSubagentRegistrySweeper },
|
|
] = await Promise.all([
|
|
import("../src/agents/subagents/registry/subagent-registry-memory.js"),
|
|
import("../src/agents/subagents/registry/subagent-registry-sweeper.js"),
|
|
]);
|
|
const now = Date.now();
|
|
runs.clear();
|
|
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: () => {},
|
|
discardTerminalDelivery: () => {},
|
|
shouldEmitEndedHookForRun: () => false,
|
|
emitSubagentEndedHookForRun: async () => {},
|
|
callGateway: (async <T>() => {
|
|
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: getSubagentRunsForChildSession,
|
|
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();
|
|
runs.clear();
|
|
}
|
|
}
|
|
|
|
async function runDedupeSample(childCount: number): Promise<Sample> {
|
|
const { dedupeLatestChildCompletionRows } =
|
|
await import("../src/agents/subagents/announce/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<WorkerResult> {
|
|
const timingsMs: number[] = [];
|
|
let invariant: Record<string, number | boolean> = {};
|
|
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<void> {
|
|
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 toErrorObject(failure, "Agent concurrency benchmark failed");
|
|
}
|
|
if (!result) {
|
|
throw new Error("benchmark worker completed without a result");
|
|
}
|
|
process.stdout.write(`${WORKER_RESULT_SENTINEL}${JSON.stringify(result)}\n`);
|
|
}
|
|
|
|
export const testing = { drainSpawnSampleRootWork };
|
|
|
|
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})`);
|
|
}
|
|
}
|
|
}
|