From d3d46725e6fe3374ae613c1a8e4b618c3069bfd9 Mon Sep 17 00:00:00 2001 From: Wibias <37517432+Wibias@users.noreply.github.com> Date: Mon, 17 Aug 2026 09:01:22 +0200 Subject: [PATCH 01/11] test(memory): add deterministic soak probe helpers --- scripts/memory-recall-soak-lib.ts | 159 ++++++++++++++++++++++++++++++ 1 file changed, 159 insertions(+) create mode 100644 scripts/memory-recall-soak-lib.ts diff --git a/scripts/memory-recall-soak-lib.ts b/scripts/memory-recall-soak-lib.ts new file mode 100644 index 0000000000..8cc2f4998d --- /dev/null +++ b/scripts/memory-recall-soak-lib.ts @@ -0,0 +1,159 @@ +export interface MemoryRecallSoakOptions { + seed: number; + sustainedSessions: number; + sustainedRounds: number; + sustainedWaves: number; + burstSessions: number; + burstRounds: number; + faultSessions: number; + slowConsumerPercent: number; + cancelPercent: number; + idleDeadlineMs: number; + sampleIntervalMs: number; +} + +export const DEFAULT_MEMORY_RECALL_SOAK_OPTIONS: MemoryRecallSoakOptions = { + seed: 820_001, + sustainedSessions: 32, + sustainedRounds: 10, + sustainedWaves: 3, + burstSessions: 64, + burstRounds: 1, + faultSessions: 32, + slowConsumerPercent: 25, + cancelPercent: 25, + idleDeadlineMs: 30_000, + sampleIntervalMs: 100, +}; + +const QUICK_MEMORY_RECALL_SOAK_OPTIONS: MemoryRecallSoakOptions = { + ...DEFAULT_MEMORY_RECALL_SOAK_OPTIONS, + sustainedSessions: 4, + sustainedRounds: 2, + sustainedWaves: 2, + burstSessions: 8, + faultSessions: 8, + idleDeadlineMs: 10_000, + sampleIntervalMs: 50, +}; + +const INTEGER_FLAGS: ReadonlyArray<{ + flag: string; + key: keyof MemoryRecallSoakOptions; + min: number; + max: number; +}> = [ + { flag: "--seed", key: "seed", min: 0, max: 0xffff_ffff }, + { flag: "--sessions", key: "sustainedSessions", min: 1, max: 96 }, + { flag: "--rounds", key: "sustainedRounds", min: 1, max: 100 }, + { flag: "--waves", key: "sustainedWaves", min: 2, max: 20 }, + { flag: "--burst-sessions", key: "burstSessions", min: 1, max: 96 }, + { flag: "--burst-rounds", key: "burstRounds", min: 1, max: 20 }, + { flag: "--fault-sessions", key: "faultSessions", min: 0, max: 96 }, + { flag: "--slow-percent", key: "slowConsumerPercent", min: 0, max: 100 }, + { flag: "--cancel-percent", key: "cancelPercent", min: 0, max: 100 }, + { flag: "--idle-deadline-ms", key: "idleDeadlineMs", min: 1_000, max: 120_000 }, + { flag: "--sample-interval-ms", key: "sampleIntervalMs", min: 25, max: 5_000 }, +]; + +export function memoryRecallSoakUsage(): string { + return [ + "Usage: bun scripts/memory-recall-soak.ts [options]", + "", + "Offline #820 acceptance/profiling probe. It starts an isolated OpenCodex child", + "against a local mock provider and emits payload-free JSON metrics.", + "", + "Options:", + " --quick Small deterministic smoke profile", + " --seed N Deterministic workload seed", + " --sessions N Sustained independent sessions (default 32)", + " --rounds N Recall rounds per sustained session (default 10)", + " --waves N Identical sustained waves (default 3)", + " --burst-sessions N Independent burst sessions (default 64)", + " --burst-rounds N Recall rounds in the burst (default 1)", + " --fault-sessions N Fault/cancel probe sessions (default 32)", + " --slow-percent N Slow-consumer share, 0..100 (default 25)", + " --cancel-percent N Fault-wave cancellation share, 0..100 (default 25)", + " --idle-deadline-ms N Cleanup invariant deadline (default 30000)", + " --sample-interval-ms N Child-memory sample cadence (default 100)", + " --help Show this help", + ].join("\n"); +} + +export function parseMemoryRecallSoakOptions(args: readonly string[]): MemoryRecallSoakOptions { + const options = args.includes("--quick") + ? { ...QUICK_MEMORY_RECALL_SOAK_OPTIONS } + : { ...DEFAULT_MEMORY_RECALL_SOAK_OPTIONS }; + const knownFlags = new Set(["--quick", "--help", ...INTEGER_FLAGS.map(row => row.flag)]); + + for (let index = 0; index < args.length; index++) { + const arg = args[index]; + if (!arg.startsWith("--")) throw new Error(`unexpected positional argument: ${arg}`); + if (!knownFlags.has(arg)) throw new Error(`unknown option: ${arg}`); + if (arg === "--quick" || arg === "--help") continue; + + const spec = INTEGER_FLAGS.find(row => row.flag === arg); + if (!spec) continue; + const raw = args[index + 1]; + if (raw === undefined || raw.startsWith("--")) throw new Error(`${arg} requires an integer value`); + const value = Number(raw); + if (!Number.isSafeInteger(value) || value < spec.min || value > spec.max) { + throw new Error(`${arg} must be an integer in ${spec.min}..${spec.max}`); + } + options[spec.key] = value; + index += 1; + } + + return options; +} + +export function mulberry32(seed: number): () => number { + let state = seed >>> 0; + return () => { + state |= 0; + state = (state + 0x6D2B79F5) | 0; + let value = Math.imul(state ^ (state >>> 15), 1 | state); + value = (value + Math.imul(value ^ (value >>> 7), 61 | value)) ^ value; + return ((value ^ (value >>> 14)) >>> 0) / 4_294_967_296; + }; +} + +export function stableHash(text: string, seed = 0): number { + let hash = (0x811c9dc5 ^ (seed >>> 0)) >>> 0; + for (let index = 0; index < text.length; index++) { + hash ^= text.charCodeAt(index); + hash = Math.imul(hash, 0x01000193) >>> 0; + } + return hash >>> 0; +} + +export function deterministicToolCount(sessionId: string, round: number, seed: number): number { + return 1 + (stableHash(`${sessionId}:${round}`, seed) % 8); +} + +export function deterministicPercent(sessionId: string, salt: string, seed: number): number { + return stableHash(`${salt}:${sessionId}`, seed) % 100; +} + +export function linearSlope(values: readonly number[]): number | null { + if (values.length < 2) return null; + const count = values.length; + const meanX = (count - 1) / 2; + const meanY = values.reduce((sum, value) => sum + value, 0) / count; + let numerator = 0; + let denominator = 0; + for (let index = 0; index < count; index++) { + const dx = index - meanX; + numerator += dx * (values[index] - meanY); + denominator += dx * dx; + } + return denominator === 0 ? null : numerator / denominator; +} + +export function maxFinite(values: readonly number[]): number | null { + let maximum = Number.NEGATIVE_INFINITY; + for (const value of values) { + if (Number.isFinite(value) && value > maximum) maximum = value; + } + return maximum === Number.NEGATIVE_INFINITY ? null : maximum; +} From 12b21712af5ee938670053d60e339e5296aa20d4 Mon Sep 17 00:00:00 2001 From: Wibias <37517432+Wibias@users.noreply.github.com> Date: Mon, 17 Aug 2026 09:01:40 +0200 Subject: [PATCH 02/11] test(memory): isolate proxy metrics in soak child --- scripts/memory-recall-soak-child.ts | 131 ++++++++++++++++++++++++++++ 1 file changed, 131 insertions(+) create mode 100644 scripts/memory-recall-soak-child.ts diff --git a/scripts/memory-recall-soak-child.ts b/scripts/memory-recall-soak-child.ts new file mode 100644 index 0000000000..d916bef208 --- /dev/null +++ b/scripts/memory-recall-soak-child.ts @@ -0,0 +1,131 @@ +/** + * #820 memory-recall soak probe child. + * + * The parent owns the mock provider and clients. This child owns only the real + * OpenCodex proxy plus a loopback-only control listener for payload-free scalar + * metrics, which keeps process RSS attributable to the proxy rather than the + * load generator. This is an offline probe, not a CI test or production route. + */ +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { OcxConfig } from "../src/types"; + +function optionValue(name: string): string | null { + const index = Bun.argv.indexOf(name); + if (index === -1 || index + 1 >= Bun.argv.length) return null; + return Bun.argv[index + 1] ?? null; +} + +const upstreamBaseUrl = optionValue("--upstream"); +if (!upstreamBaseUrl) { + console.error("memory-recall-soak-child: --upstream is required"); + process.exit(2); +} + +let upstreamUrl: URL; +try { + upstreamUrl = new URL(upstreamBaseUrl); +} catch { + console.error("memory-recall-soak-child: --upstream must be a URL"); + process.exit(2); +} +if (upstreamUrl.hostname !== "127.0.0.1" && upstreamUrl.hostname !== "localhost" && upstreamUrl.hostname !== "::1") { + console.error("memory-recall-soak-child: upstream must be loopback"); + process.exit(2); +} + +const home = mkdtempSync(join(tmpdir(), "ocx-memory-recall-soak-")); +process.env.OPENCODEX_HOME = home; + +const [configModule, serverModule, memoryModule, lifecycleModule, relayModule, responseStateModule] = await Promise.all([ + import("../src/config"), + import("../src/server"), + import("../src/lib/app-owned-memory"), + import("../src/server/lifecycle"), + import("../src/server/relay"), + import("../src/responses/state"), +]); + +const config: OcxConfig = { + port: 0, + defaultProvider: "mock", + providers: { + mock: { + adapter: "openai-chat", + baseUrl: `${upstreamUrl.toString().replace(/\/$/, "")}/v1`, + apiKey: "memory-soak-local-only", + allowPrivateNetwork: true, + }, + }, +}; +configModule.saveConfig(config); +const proxy = serverModule.startServer(0); + +function metrics() { + const usage = process.memoryUsage(); + return { + atMs: Date.now(), + pid: process.pid, + platform: process.platform, + bunVersion: Bun.version, + bunRevision: Bun.revision, + uptimeSeconds: process.uptime(), + rss: usage.rss, + heapUsed: usage.heapUsed, + heapTotal: usage.heapTotal, + external: usage.external, + arrayBuffers: usage.arrayBuffers, + activeTurnCount: lifecycleModule.getActiveTurnCount(), + appOwnedBytes: memoryModule.appOwnedBytesSnapshot(), + inspectionCounters: relayModule.getInspectionCounters(), + responseState: responseStateModule.responseStateMetrics(), + }; +} + +let closing = false; +let control: ReturnType; + +async function closeAndExit(code: number): Promise { + if (closing) { + await Bun.sleep(25); + process.exit(code); + } + closing = true; + try { await proxy.stop(true); } catch { /* best-effort probe teardown */ } + try { control?.stop(true); } catch { /* best-effort probe teardown */ } + try { rmSync(home, { recursive: true, force: true }); } catch { /* temp cleanup only */ } + process.exit(code); +} + +control = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + async fetch(req) { + const url = new URL(req.url); + if (url.pathname === "/metrics" && req.method === "GET") { + return Response.json(metrics(), { + headers: { "cache-control": "no-store" }, + }); + } + if (url.pathname === "/shutdown" && req.method === "POST") { + setTimeout(() => { void closeAndExit(0); }, 0); + return Response.json({ ok: true }); + } + return new Response("not found", { status: 404 }); + }, +}); + +console.log(JSON.stringify({ + type: "ready", + proxyUrl: proxy.url.toString(), + controlUrl: control.url.toString(), + pid: process.pid, + platform: process.platform, + bunVersion: Bun.version, + bunRevision: Bun.revision, +})); + +for (const signal of ["SIGINT", "SIGTERM"] as const) { + process.on(signal, () => { void closeAndExit(128); }); +} From 2ef5b1e4aa5854740f9ac173bd75e786f9cae5fa Mon Sep 17 00:00:00 2001 From: Wibias <37517432+Wibias@users.noreply.github.com> Date: Mon, 17 Aug 2026 09:03:01 +0200 Subject: [PATCH 03/11] test(memory): add concurrent recall soak probe --- scripts/memory-recall-soak.ts | 722 ++++++++++++++++++++++++++++++++++ 1 file changed, 722 insertions(+) create mode 100644 scripts/memory-recall-soak.ts diff --git a/scripts/memory-recall-soak.ts b/scripts/memory-recall-soak.ts new file mode 100644 index 0000000000..acf67a9d2a --- /dev/null +++ b/scripts/memory-recall-soak.ts @@ -0,0 +1,722 @@ +/** + * #820 concurrent recall acceptance/profiling probe. + * + * This is intentionally an offline probe, not a normal test/CI job. The parent + * owns the mock provider and load generator; a child process owns the real + * OpenCodex proxy so RSS samples do not include the clients that create load. + * + * Full defaults exercise 32 sustained independent sessions for 10 recall rounds, + * three identical waves, a 64-session burst, slow consumers, and a fault wave. + * RSS is evidence only: cleanup assertions are made against OpenCodex-owned + * counters, while RSS slope is reported for leak-vs-allocator-retention analysis. + */ +import { + deterministicPercent, + deterministicToolCount, + linearSlope, + maxFinite, + memoryRecallSoakUsage, + parseMemoryRecallSoakOptions, + stableHash, + type MemoryRecallSoakOptions, +} from "./memory-recall-soak-lib"; + +interface ChildReady { + type: "ready"; + proxyUrl: string; + controlUrl: string; + pid: number; + platform: string; + bunVersion: string; + bunRevision: string; +} + +interface AppOwnedSnapshot { + retainedBytes: number; + evictableBytes: number; + pinnedBytes: number; + overBudgetBytes: number; + observedInFlight: Record; +} + +interface ProbeMetrics { + atMs: number; + rss: number; + heapUsed: number; + heapTotal: number; + external: number; + arrayBuffers: number; + activeTurnCount: number; + appOwnedBytes: AppOwnedSnapshot; + inspectionCounters: Record; + responseState: Record; +} + +interface WaveResult { + name: string; + sessions: number; + rounds: number; + completedSessions: number; + failedSessions: number; + requestCount: number; + toolCallCount: number; + peak: ProbeMetrics; + idle: ProbeMetrics; + durationMs: number; +} + +type FaultKind = "cancel" | "http_429" | "http_503" | "pre_first_byte_stream_error"; + +type CompletedResponse = { + id?: string; + status?: string; + output?: Array>; +}; + +type SessionResult = { + requests: number; + toolCalls: number; +}; + +const encoder = new TextEncoder(); +const args = Bun.argv.slice(2); +if (args.includes("--help")) { + console.log(memoryRecallSoakUsage()); + process.exit(0); +} + +let options: MemoryRecallSoakOptions; +try { + options = parseMemoryRecallSoakOptions(args); +} catch (error) { + console.error(error instanceof Error ? error.message : String(error)); + console.error(memoryRecallSoakUsage()); + process.exit(2); +} + +function emit(event: Record): void { + console.log(JSON.stringify(event)); +} + +function boundedTail(current: string, chunk: string): string { + return (current + chunk).slice(-16_384); +} + +function sessionMarker(sessionId: string): string { + return `OCX_MEMORY_SOAK_SESSION_${sessionId}`; +} + +function roundMarker(round: number): string { + return `OCX_MEMORY_SOAK_ROUND_${round}_DONE`; +} + +function extractSessionId(body: unknown): string { + const text = JSON.stringify(body); + const match = text.match(/OCX_MEMORY_SOAK_SESSION_([A-Za-z0-9_-]{1,80})/); + return match?.[1] ?? "unknown"; +} + +function extractRound(body: unknown): number { + const text = JSON.stringify(body); + let maximum = -1; + for (const match of text.matchAll(/OCX_MEMORY_SOAK_ROUND_(\d+)_DONE/g)) { + const value = Number(match[1]); + if (Number.isSafeInteger(value)) maximum = Math.max(maximum, value); + } + return maximum + 1; +} + +function advertisedToolNames(body: Record): string[] { + const tools = Array.isArray(body.tools) ? body.tools : []; + const names: string[] = []; + for (const tool of tools) { + if (!tool || typeof tool !== "object" || Array.isArray(tool)) continue; + const fn = (tool as Record).function; + if (!fn || typeof fn !== "object" || Array.isArray(fn)) continue; + const name = (fn as Record).name; + if (typeof name === "string" && name) names.push(name); + } + return names; +} + +function orderedToolNames(names: readonly string[]): string[] { + const ordinary = names.filter(name => !/apply_patch|tool_search/i.test(name)); + const special = names.filter(name => /apply_patch|tool_search/i.test(name)); + return [...ordinary, ...special]; +} + +function toolArguments(name: string, sessionId: string, round: number, index: number): string { + if (/apply_patch/i.test(name)) { + return JSON.stringify({ input: `*** Begin Patch\n*** Add File: soak-${round}-${index}.txt\n+probe\n*** End Patch` }); + } + if (/tool_search/i.test(name)) { + return JSON.stringify({ query: `probe ${round} ${index}`, limit: 3 }); + } + return JSON.stringify({ session: sessionId, round, index, payload: "x".repeat(256 + index * 17) }); +} + +function faultKind(sessionId: string): FaultKind { + const bucket = deterministicPercent(sessionId, "fault", options.seed); + if (bucket < options.cancelPercent) return "cancel"; + const remainder = stableHash(`fault-kind:${sessionId}`, options.seed) % 3; + return remainder === 0 ? "http_429" : remainder === 1 ? "http_503" : "pre_first_byte_stream_error"; +} + +function streamFrames(frames: readonly string[], jitterSeed: number): ReadableStream { + let index = 0; + return new ReadableStream({ + async pull(controller) { + if (index >= frames.length) { + controller.close(); + return; + } + if (((jitterSeed + index) & 3) === 0) await Bun.sleep(1); + controller.enqueue(encoder.encode(frames[index++])); + }, + }); +} + +function buildToolFrames(body: Record): string[] { + const sessionId = extractSessionId(body); + const round = extractRound(body); + const names = orderedToolNames(advertisedToolNames(body)); + if (names.length === 0) throw new Error("mock upstream received no callable tools"); + const count = deterministicToolCount(sessionId, round, options.seed); + const selected = Array.from({ length: count }, (_, index) => names[index % names.length]); + const starts: string[] = []; + const finishes: string[] = []; + + for (let index = 0; index < selected.length; index++) { + const name = selected[index]; + const args = toolArguments(name, sessionId, round, index); + const split = Math.max(1, Math.floor(args.length / 2)); + starts.push(`data: ${JSON.stringify({ + choices: [{ + index: 0, + delta: { + tool_calls: [{ + index, + id: `call_${stableHash(`${sessionId}:${round}:${index}`, options.seed).toString(16)}`, + type: "function", + function: { name, arguments: args.slice(0, split) }, + }], + }, + }], + })}\n\n`); + finishes.unshift(`data: ${JSON.stringify({ + choices: [{ + index: 0, + delta: { tool_calls: [{ index, function: { arguments: args.slice(split) } }] }, + }], + })}\n\n`); + } + return [ + ...starts, + ...finishes, + `data: ${JSON.stringify({ choices: [{ index: 0, delta: {}, finish_reason: "tool_calls" }] })}\n\n`, + "data: [DONE]\n\n", + ]; +} + +const upstream = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + async fetch(req) { + const url = new URL(req.url); + if (!url.pathname.endsWith("/chat/completions")) { + return Response.json({ error: { message: "unexpected mock-provider path" } }, { status: 404 }); + } + let body: Record; + try { + body = await req.json() as Record; + } catch { + return Response.json({ error: { message: "invalid mock-provider body" } }, { status: 400 }); + } + const sessionId = extractSessionId(body); + if (sessionId.startsWith("fault-")) { + const kind = faultKind(sessionId); + if (kind === "http_429") { + return Response.json({ error: { message: "synthetic rate limit" } }, { status: 429 }); + } + if (kind === "http_503") { + return Response.json({ error: { message: "synthetic unavailable" } }, { status: 503 }); + } + if (kind === "pre_first_byte_stream_error") { + return new Response(new ReadableStream({ + pull(controller) { + controller.error(new Error("synthetic pre-first-byte stream failure")); + }, + }), { headers: { "content-type": "text/event-stream" } }); + } + } + const frames = buildToolFrames(body); + return new Response(streamFrames(frames, stableHash(sessionId, options.seed)), { + headers: { "content-type": "text/event-stream" }, + }); + }, +}); + +let childStdoutTail = ""; +let childStderrTail = ""; +const child = Bun.spawn({ + cmd: [ + process.execPath, + `${import.meta.dir}/memory-recall-soak-child.ts`, + "--upstream", + upstream.url.toString().replace(/\/$/, ""), + ], + cwd: `${import.meta.dir}/..`, + stdout: "pipe", + stderr: "pipe", +}); + +let resolveReady: ((value: ChildReady) => void) | null = null; +let rejectReady: ((error: Error) => void) | null = null; +const readyPromise = new Promise((resolve, reject) => { + resolveReady = resolve; + rejectReady = reject; +}); + +async function consumeChildStdout(): Promise { + const reader = child.stdout.getReader(); + const decoder = new TextDecoder(); + let pending = ""; + while (true) { + const { done, value } = await reader.read(); + if (done) break; + const text = decoder.decode(value, { stream: true }); + childStdoutTail = boundedTail(childStdoutTail, text); + pending += text; + for (;;) { + const newline = pending.indexOf("\n"); + if (newline < 0) break; + const line = pending.slice(0, newline).trim(); + pending = pending.slice(newline + 1); + if (!line) continue; + try { + const event = JSON.parse(line) as Partial; + if (event.type === "ready" && typeof event.proxyUrl === "string" && typeof event.controlUrl === "string") { + resolveReady?.(event as ChildReady); + resolveReady = null; + rejectReady = null; + } + } catch { /* bounded tail remains available on failure */ } + } + } +} + +async function consumeChildStderr(): Promise { + const reader = child.stderr.getReader(); + const decoder = new TextDecoder(); + while (true) { + const { done, value } = await reader.read(); + if (done) break; + childStderrTail = boundedTail(childStderrTail, decoder.decode(value, { stream: true })); + } +} + +void consumeChildStdout().catch(error => rejectReady?.(error instanceof Error ? error : new Error(String(error)))); +void consumeChildStderr(); +void child.exited.then(code => { + if (resolveReady) rejectReady?.(new Error(`proxy child exited before readiness with code ${code}`)); +}); + +async function withTimeout(promise: Promise, ms: number, label: string): Promise { + let timer: ReturnType | undefined; + try { + return await Promise.race([ + promise, + new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(`${label} timed out after ${ms} ms`)), ms); + }), + ]); + } finally { + if (timer) clearTimeout(timer); + } +} + +const ready = await withTimeout(readyPromise, 30_000, "proxy child readiness"); +const proxyBase = new URL(ready.proxyUrl); +const controlBase = new URL(ready.controlUrl); + +async function sampleMetrics(): Promise { + const response = await fetch(new URL("/metrics", controlBase), { cache: "no-store" }); + if (!response.ok) throw new Error(`metrics endpoint returned ${response.status}`); + return await response.json() as ProbeMetrics; +} + +function idleInvariant(metrics: ProbeMetrics): boolean { + if (metrics.activeTurnCount !== 0) return false; + return Object.values(metrics.appOwnedBytes.observedInFlight).every(row => row.currentBytes === 0 && row.active === 0); +} + +async function waitForIdle(): Promise { + const deadline = Date.now() + options.idleDeadlineMs; + let latest = await sampleMetrics(); + while (!idleInvariant(latest) && Date.now() < deadline) { + await Bun.sleep(Math.min(100, options.sampleIntervalMs)); + latest = await sampleMetrics(); + } + if (!idleInvariant(latest)) { + throw new Error(`proxy did not return to app-owned idle invariants within ${options.idleDeadlineMs} ms`); + } + return latest; +} + +function tools(): Array> { + const namespaceTools = Array.from({ length: 6 }, (_, index) => ({ + type: "function", + name: `read_${index}`, + description: `Synthetic namespace tool ${index}`, + parameters: { + type: "object", + properties: { + session: { type: "string" }, + round: { type: "integer" }, + index: { type: "integer" }, + payload: { type: "string" }, + }, + required: ["session", "round", "index", "payload"], + additionalProperties: false, + }, + })); + return [ + { + type: "namespace", + name: "workspace", + description: "Synthetic MCP-style namespace", + tools: namespaceTools, + }, + { type: "custom", name: "apply_patch", description: "Synthetic freeform patch tool" }, + { + type: "tool_search", + execution: "client", + description: "Synthetic deferred-tool search", + parameters: { + type: "object", + properties: { query: { type: "string" }, limit: { type: "integer" } }, + required: ["query"], + additionalProperties: false, + }, + }, + ]; +} + +function toolSearchResultTools(): Array> { + return [{ + type: "namespace", + name: "deferred", + description: "Synthetic deferred namespace", + tools: [{ + type: "function", + name: "deferred_read", + description: "Synthetic deferred read", + defer_loading: true, + parameters: { type: "object", properties: {}, additionalProperties: false }, + }], + }]; +} + +async function readResponseText(response: Response, slow: boolean, signal?: AbortSignal): Promise { + if (!response.body) return ""; + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + let text = ""; + try { + while (true) { + if (signal?.aborted) throw signal.reason ?? new Error("aborted"); + const { done, value } = await reader.read(); + if (done) break; + text += decoder.decode(value, { stream: true }); + if (slow) await Bun.sleep(4); + } + text += decoder.decode(); + return text; + } finally { + reader.releaseLock(); + } +} + +function completedResponseFromSse(text: string): CompletedResponse | null { + let completed: CompletedResponse | null = null; + for (const line of text.split("\n")) { + if (!line.startsWith("data: ")) continue; + const data = line.slice(6).trim(); + if (!data || data === "[DONE]") continue; + try { + const event = JSON.parse(data) as { type?: string; response?: CompletedResponse }; + if (event.type === "response.completed" && event.response) completed = event.response; + } catch { /* malformed frames are validated by the absence of a completed response */ } + } + return completed; +} + +function callableItems(output: readonly Record[]): Array> { + return output.filter(item => item.type === "function_call" || item.type === "custom_tool_call" || item.type === "tool_search_call"); +} + +function appendRecallOutput(input: Array>, item: Record, round: number): void { + input.push(item); + const callId = typeof item.call_id === "string" ? item.call_id : null; + if (!callId) throw new Error(`completed ${String(item.type)} item omitted call_id`); + const marker = `${roundMarker(round)} ok`; + if (item.type === "custom_tool_call") { + input.push({ type: "custom_tool_call_output", call_id: callId, output: marker }); + return; + } + if (item.type === "tool_search_call") { + input.push({ + type: "tool_search_output", + call_id: callId, + status: "completed", + execution: "client", + tools: toolSearchResultTools(), + }); + return; + } + input.push({ type: "function_call_output", call_id: callId, output: marker }); +} + +async function runSession(sessionId: string, rounds: number, slow: boolean): Promise { + const input: Array> = [{ + type: "message", + role: "user", + content: [{ type: "input_text", text: `${sessionMarker(sessionId)} run the synthetic tools` }], + }]; + let toolCalls = 0; + + for (let round = 0; round < rounds; round++) { + const response = await fetch(new URL("/v1/responses", proxyBase), { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model: "mock/test-model", + stream: true, + store: false, + input, + tools: tools(), + tool_choice: "auto", + }), + }); + if (response.status !== 200) { + const detail = (await response.text()).slice(0, 240); + throw new Error(`round ${round} returned HTTP ${response.status}: ${detail}`); + } + const text = await readResponseText(response, slow); + const completed = completedResponseFromSse(text); + if (!completed || completed.status !== "completed" || !Array.isArray(completed.output)) { + throw new Error(`round ${round} did not produce response.completed`); + } + const calls = callableItems(completed.output); + const expected = deterministicToolCount(sessionId, round, options.seed); + if (calls.length !== expected) { + throw new Error(`round ${round} completed ${calls.length} tool calls; expected ${expected}`); + } + const callIds = new Set(calls.map(item => item.call_id)); + if (callIds.size !== calls.length || callIds.has(undefined)) { + throw new Error(`round ${round} produced missing or duplicate call ids`); + } + toolCalls += calls.length; + for (const item of calls) appendRecallOutput(input, item, round); + } + return { requests: rounds, toolCalls }; +} + +function peakMetrics(samples: readonly ProbeMetrics[]): ProbeMetrics { + if (samples.length === 0) throw new Error("wave collected no memory samples"); + return samples.reduce((peak, sample) => sample.rss > peak.rss ? sample : peak); +} + +async function runWave(name: string, sessions: number, rounds: number): Promise { + const started = Date.now(); + const samples: ProbeMetrics[] = []; + let monitoring = true; + const monitor = (async () => { + while (monitoring) { + try { samples.push(await sampleMetrics()); } catch { /* main wave outcome remains authoritative */ } + if (monitoring) await Bun.sleep(options.sampleIntervalMs); + } + })(); + + const settled = await Promise.allSettled(Array.from({ length: sessions }, (_, index) => { + const sessionId = `${name}-${index}`; + const slow = deterministicPercent(sessionId, "slow", options.seed) < options.slowConsumerPercent; + return runSession(sessionId, rounds, slow); + })); + monitoring = false; + await monitor; + samples.push(await sampleMetrics()); + + const failures = settled.filter(result => result.status === "rejected"); + const successes = settled.filter((result): result is PromiseFulfilledResult => result.status === "fulfilled"); + const idle = await waitForIdle(); + samples.push(idle); + const result: WaveResult = { + name, + sessions, + rounds, + completedSessions: successes.length, + failedSessions: failures.length, + requestCount: successes.reduce((sum, result) => sum + result.value.requests, 0), + toolCallCount: successes.reduce((sum, result) => sum + result.value.toolCalls, 0), + peak: peakMetrics(samples), + idle, + durationMs: Date.now() - started, + }; + emit({ + type: "WAVE", + name, + sessions, + rounds, + completedSessions: result.completedSessions, + failedSessions: result.failedSessions, + requestCount: result.requestCount, + toolCallCount: result.toolCallCount, + peakRss: result.peak.rss, + idleRss: idle.rss, + idleRetainedBytes: idle.appOwnedBytes.retainedBytes, + durationMs: result.durationMs, + firstFailure: failures[0]?.status === "rejected" + ? String(failures[0].reason instanceof Error ? failures[0].reason.message : failures[0].reason).slice(0, 240) + : undefined, + }); + return result; +} + +async function cancelOneResponse(sessionId: string): Promise { + const controller = new AbortController(); + const response = await fetch(new URL("/v1/responses", proxyBase), { + method: "POST", + signal: controller.signal, + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model: "mock/test-model", + stream: true, + store: false, + input: sessionMarker(sessionId), + tools: tools(), + }), + }); + const reader = response.body?.getReader(); + if (!reader) throw new Error("cancel probe response had no body"); + try { + await reader.read(); + controller.abort(new Error("synthetic client cancel")); + try { await reader.read(); } catch { /* cancellation is the expected outcome */ } + } finally { + try { await reader.cancel(); } catch { /* already aborted */ } + } + return "cancelled"; +} + +async function runFaultSession(index: number): Promise<{ kind: FaultKind; outcome: string }> { + const sessionId = `fault-${index}`; + const kind = faultKind(sessionId); + if (kind === "cancel") return { kind, outcome: await cancelOneResponse(sessionId) }; + + try { + const response = await fetch(new URL("/v1/responses", proxyBase), { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model: "mock/test-model", + stream: true, + store: false, + input: sessionMarker(sessionId), + tools: tools(), + }), + }); + let text = ""; + try { text = await readResponseText(response, false); } catch { return { kind, outcome: "body-read-error" }; } + const completed = completedResponseFromSse(text); + return { kind, outcome: completed?.status ? `terminal-${completed.status}` : `http-${response.status}` }; + } catch { + return { kind, outcome: "fetch-error" }; + } +} + +async function runFaultWave(): Promise> { + if (options.faultSessions === 0) return {}; + const results = await Promise.all(Array.from({ length: options.faultSessions }, (_, index) => runFaultSession(index))); + const counts: Record = {}; + for (const result of results) { + const key = `${result.kind}:${result.outcome}`; + counts[key] = (counts[key] ?? 0) + 1; + } + await waitForIdle(); + emit({ type: "FAULT_WAVE", sessions: options.faultSessions, outcomes: counts }); + return counts; +} + +let exitCode = 0; +try { + const initial = await waitForIdle(); + emit({ + type: "START", + seed: options.seed, + algorithm: "stable-fnv1a32", + child: ready, + initialRss: initial.rss, + initialRetainedBytes: initial.appOwnedBytes.retainedBytes, + options, + }); + + const sustained: WaveResult[] = []; + for (let wave = 0; wave < options.sustainedWaves; wave++) { + sustained.push(await runWave(`sustained-${wave}`, options.sustainedSessions, options.sustainedRounds)); + } + const burst = await runWave("burst", options.burstSessions, options.burstRounds); + const faultOutcomes = await runFaultWave(); + const final = await waitForIdle(); + + const failedBaselineSessions = sustained.reduce((sum, wave) => sum + wave.failedSessions, 0) + burst.failedSessions; + const idleRss = sustained.map(wave => wave.idle.rss); + const idleRetained = sustained.map(wave => wave.idle.appOwnedBytes.retainedBytes); + const peakRss = maxFinite([...sustained.map(wave => wave.peak.rss), burst.peak.rss]); + const summary = { + type: "SUMMARY", + outcome: failedBaselineSessions === 0 ? "PASS" : "FAIL", + seed: options.seed, + algorithm: "stable-fnv1a32", + platform: ready.platform, + bunVersion: ready.bunVersion, + bunRevision: ready.bunRevision, + sustainedWaves: options.sustainedWaves, + failedBaselineSessions, + peakRss, + initialRss: initial.rss, + finalRss: final.rss, + rssIdleSlopeBytesPerWave: linearSlope(idleRss), + retainedIdleSlopeBytesPerWave: linearSlope(idleRetained), + finalRetainedBytes: final.appOwnedBytes.retainedBytes, + finalPinnedBytes: final.appOwnedBytes.pinnedBytes, + finalOverBudgetBytes: final.appOwnedBytes.overBudgetBytes, + finalActiveTurnCount: final.activeTurnCount, + faultOutcomes, + note: "RSS slope is profiling evidence only; PASS is based on protocol completion and app-owned cleanup invariants.", + }; + emit(summary); + if (failedBaselineSessions !== 0) exitCode = 1; +} catch (error) { + exitCode = 1; + emit({ + type: "SUMMARY", + outcome: "FAIL", + classification: "probe-error", + seed: options.seed, + failure: error instanceof Error ? error.message : String(error), + childStdoutTail, + childStderrTail, + }); +} finally { + try { + await fetch(new URL("/shutdown", controlBase), { method: "POST" }); + } catch { /* child may already have exited */ } + upstream.stop(true); + const childExit = await Promise.race([child.exited, Bun.sleep(2_000).then(() => null)]); + if (childExit === null) { + try { child.kill("SIGKILL"); } catch { /* already exited */ } + } +} + +process.exit(exitCode); From e58ce9e8bf0fb453c2a714aa66f5bac8f4e1d9bb Mon Sep 17 00:00:00 2001 From: Wibias <37517432+Wibias@users.noreply.github.com> Date: Mon, 17 Aug 2026 09:03:17 +0200 Subject: [PATCH 04/11] test(memory): cover soak probe determinism --- tests/memory-recall-soak.test.ts | 71 ++++++++++++++++++++++++++++++++ 1 file changed, 71 insertions(+) create mode 100644 tests/memory-recall-soak.test.ts diff --git a/tests/memory-recall-soak.test.ts b/tests/memory-recall-soak.test.ts new file mode 100644 index 0000000000..b97d0e3fc5 --- /dev/null +++ b/tests/memory-recall-soak.test.ts @@ -0,0 +1,71 @@ +import { describe, expect, test } from "bun:test"; +import { + DEFAULT_MEMORY_RECALL_SOAK_OPTIONS, + deterministicPercent, + deterministicToolCount, + linearSlope, + maxFinite, + mulberry32, + parseMemoryRecallSoakOptions, + stableHash, +} from "../scripts/memory-recall-soak-lib"; + +describe("#820 memory recall soak probe helpers", () => { + test("full defaults preserve the acceptance workload contract", () => { + expect(parseMemoryRecallSoakOptions([])).toEqual(DEFAULT_MEMORY_RECALL_SOAK_OPTIONS); + expect(DEFAULT_MEMORY_RECALL_SOAK_OPTIONS).toMatchObject({ + sustainedSessions: 32, + sustainedRounds: 10, + sustainedWaves: 3, + burstSessions: 64, + slowConsumerPercent: 25, + cancelPercent: 25, + }); + }); + + test("quick mode stays bounded and explicit overrides win", () => { + expect(parseMemoryRecallSoakOptions([ + "--quick", + "--sessions", "6", + "--rounds", "3", + "--fault-sessions", "0", + ])).toMatchObject({ + sustainedSessions: 6, + sustainedRounds: 3, + sustainedWaves: 2, + burstSessions: 8, + faultSessions: 0, + }); + }); + + test("invalid numeric and unknown options fail closed", () => { + expect(() => parseMemoryRecallSoakOptions(["--sessions", "0"])).toThrow(); + expect(() => parseMemoryRecallSoakOptions(["--sessions", "97"])).toThrow(); + expect(() => parseMemoryRecallSoakOptions(["--slow-percent", "101"])).toThrow(); + expect(() => parseMemoryRecallSoakOptions(["--unknown"])).toThrow(); + expect(() => parseMemoryRecallSoakOptions(["positional"])).toThrow(); + }); + + test("seeded workload decisions are reproducible and remain in bounds", () => { + const first = mulberry32(820_001); + const second = mulberry32(820_001); + expect(Array.from({ length: 8 }, () => first())).toEqual(Array.from({ length: 8 }, () => second())); + + for (let index = 0; index < 128; index++) { + const session = `session-${index}`; + expect(stableHash(session, 7)).toBe(stableHash(session, 7)); + expect(deterministicToolCount(session, index % 10, 7)).toBeWithin(1, 8); + expect(deterministicPercent(session, "slow", 7)).toBeWithin(0, 99); + } + }); + + test("idle-wave slope reports direction without inventing an RSS pass threshold", () => { + expect(linearSlope([])).toBeNull(); + expect(linearSlope([100])).toBeNull(); + expect(linearSlope([100, 120, 140])).toBe(20); + expect(linearSlope([140, 120, 100])).toBe(-20); + expect(linearSlope([100, 100, 100])).toBe(0); + expect(maxFinite([1, 9, 3])).toBe(9); + expect(maxFinite([])).toBeNull(); + }); +}); From c7883bb59dac6ba540d68759695d141fbf3df4b0 Mon Sep 17 00:00:00 2001 From: Wibias <37517432+Wibias@users.noreply.github.com> Date: Mon, 17 Aug 2026 09:05:12 +0200 Subject: [PATCH 05/11] test(memory): use portable numeric assertions --- tests/memory-recall-soak.test.ts | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/tests/memory-recall-soak.test.ts b/tests/memory-recall-soak.test.ts index b97d0e3fc5..f354ca2db2 100644 --- a/tests/memory-recall-soak.test.ts +++ b/tests/memory-recall-soak.test.ts @@ -54,8 +54,14 @@ describe("#820 memory recall soak probe helpers", () => { for (let index = 0; index < 128; index++) { const session = `session-${index}`; expect(stableHash(session, 7)).toBe(stableHash(session, 7)); - expect(deterministicToolCount(session, index % 10, 7)).toBeWithin(1, 8); - expect(deterministicPercent(session, "slow", 7)).toBeWithin(0, 99); + + const toolCount = deterministicToolCount(session, index % 10, 7); + expect(toolCount).toBeGreaterThanOrEqual(1); + expect(toolCount).toBeLessThanOrEqual(8); + + const percent = deterministicPercent(session, "slow", 7); + expect(percent).toBeGreaterThanOrEqual(0); + expect(percent).toBeLessThanOrEqual(99); } }); From c14d612a4f1eeb67afc51227d25f0b5671014f22 Mon Sep 17 00:00:00 2001 From: Wibias <37517432+Wibias@users.noreply.github.com> Date: Mon, 17 Aug 2026 09:05:29 +0200 Subject: [PATCH 06/11] test(memory): harden cross-platform soak child setup --- scripts/memory-recall-soak-child.ts | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/scripts/memory-recall-soak-child.ts b/scripts/memory-recall-soak-child.ts index d916bef208..e800fca86b 100644 --- a/scripts/memory-recall-soak-child.ts +++ b/scripts/memory-recall-soak-child.ts @@ -47,7 +47,7 @@ const [configModule, serverModule, memoryModule, lifecycleModule, relayModule, r import("../src/responses/state"), ]); -const config: OcxConfig = { +const config = { port: 0, defaultProvider: "mock", providers: { @@ -58,7 +58,7 @@ const config: OcxConfig = { allowPrivateNetwork: true, }, }, -}; +} as OcxConfig; configModule.saveConfig(config); const proxy = serverModule.startServer(0); @@ -84,7 +84,7 @@ function metrics() { } let closing = false; -let control: ReturnType; +let control: ReturnType | undefined; async function closeAndExit(code: number): Promise { if (closing) { From 35279cad37470481b7194077cba383dc6747fa4a Mon Sep 17 00:00:00 2001 From: Wibias <37517432+Wibias@users.noreply.github.com> Date: Mon, 17 Aug 2026 09:07:58 +0200 Subject: [PATCH 07/11] test(memory): synchronize recall waves and report pressure peaks --- scripts/memory-recall-soak.ts | 416 ++++++++++++++++++++++------------ 1 file changed, 267 insertions(+), 149 deletions(-) diff --git a/scripts/memory-recall-soak.ts b/scripts/memory-recall-soak.ts index acf67a9d2a..1c60962f2e 100644 --- a/scripts/memory-recall-soak.ts +++ b/scripts/memory-recall-soak.ts @@ -5,16 +5,17 @@ * owns the mock provider and load generator; a child process owns the real * OpenCodex proxy so RSS samples do not include the clients that create load. * - * Full defaults exercise 32 sustained independent sessions for 10 recall rounds, - * three identical waves, a 64-session burst, slow consumers, and a fault wave. - * RSS is evidence only: cleanup assertions are made against OpenCodex-owned - * counters, while RSS slope is reported for leak-vs-allocator-retention analysis. + * Full defaults exercise 32 sustained independent sessions for 10 barrier- + * synchronized recall rounds, three identical waves, a 64-session burst, slow + * consumers, parallel tool latency, and a fault wave. RSS is evidence only: + * cleanup assertions use OpenCodex-owned counters, while process-memory slopes + * are reported for leak-vs-allocator-retention analysis. */ +import { join } from "node:path"; import { deterministicPercent, deterministicToolCount, linearSlope, - maxFinite, memoryRecallSoakUsage, parseMemoryRecallSoakOptions, stableHash, @@ -52,6 +53,18 @@ interface ProbeMetrics { responseState: Record; } +interface MetricPeaks { + rss: number; + heapUsed: number; + heapTotal: number; + external: number; + arrayBuffers: number; + activeTurnCount: number; + retainedBytes: number; + observedInFlightBytes: number; + observedInFlightHighWaterBytes: number; +} + interface WaveResult { name: string; sessions: number; @@ -60,11 +73,20 @@ interface WaveResult { failedSessions: number; requestCount: number; toolCallCount: number; - peak: ProbeMetrics; + peaks: MetricPeaks; idle: ProbeMetrics; durationMs: number; } +interface SessionState { + id: string; + input: Array>; + slow: boolean; + requests: number; + toolCalls: number; + failure?: string; +} + type FaultKind = "cancel" | "http_429" | "http_503" | "pre_first_byte_stream_error"; type CompletedResponse = { @@ -73,11 +95,6 @@ type CompletedResponse = { output?: Array>; }; -type SessionResult = { - requests: number; - toolCalls: number; -}; - const encoder = new TextEncoder(); const args = Bun.argv.slice(2); if (args.includes("--help")) { @@ -98,10 +115,6 @@ function emit(event: Record): void { console.log(JSON.stringify(event)); } -function boundedTail(current: string, chunk: string): string { - return (current + chunk).slice(-16_384); -} - function sessionMarker(sessionId: string): string { return `OCX_MEMORY_SOAK_SESSION_${sessionId}`; } @@ -162,7 +175,11 @@ function faultKind(sessionId: string): FaultKind { return remainder === 0 ? "http_429" : remainder === 1 ? "http_503" : "pre_first_byte_stream_error"; } -function streamFrames(frames: readonly string[], jitterSeed: number): ReadableStream { +function streamFrames( + frames: readonly string[], + jitterSeed: number, + minimumDelayMs = 0, +): ReadableStream { let index = 0; return new ReadableStream({ async pull(controller) { @@ -170,7 +187,8 @@ function streamFrames(frames: readonly string[], jitterSeed: number): ReadableSt controller.close(); return; } - if (((jitterSeed + index) & 3) === 0) await Bun.sleep(1); + const jitterMs = ((jitterSeed + index) & 3) === 0 ? 1 : 0; + if (minimumDelayMs + jitterMs > 0) await Bun.sleep(minimumDelayMs + jitterMs); controller.enqueue(encoder.encode(frames[index++])); }, }); @@ -188,8 +206,8 @@ function buildToolFrames(body: Record): string[] { for (let index = 0; index < selected.length; index++) { const name = selected[index]; - const args = toolArguments(name, sessionId, round, index); - const split = Math.max(1, Math.floor(args.length / 2)); + const toolArgs = toolArguments(name, sessionId, round, index); + const split = Math.max(1, Math.floor(toolArgs.length / 2)); starts.push(`data: ${JSON.stringify({ choices: [{ index: 0, @@ -198,7 +216,7 @@ function buildToolFrames(body: Record): string[] { index, id: `call_${stableHash(`${sessionId}:${round}:${index}`, options.seed).toString(16)}`, type: "function", - function: { name, arguments: args.slice(0, split) }, + function: { name, arguments: toolArgs.slice(0, split) }, }], }, }], @@ -206,7 +224,7 @@ function buildToolFrames(body: Record): string[] { finishes.unshift(`data: ${JSON.stringify({ choices: [{ index: 0, - delta: { tool_calls: [{ index, function: { arguments: args.slice(split) } }] }, + delta: { tool_calls: [{ index, function: { arguments: toolArgs.slice(split) } }] }, }], })}\n\n`); } @@ -218,6 +236,21 @@ function buildToolFrames(body: Record): string[] { ]; } +function buildCancelFrames(sessionId: string): string[] { + const frames = Array.from({ length: 128 }, (_, index) => `data: ${JSON.stringify({ + choices: [{ + index: 0, + delta: { + ...(index === 0 ? { role: "assistant" } : {}), + content: `${sessionId}:${index}:${"x".repeat(2 * 1024)}`, + }, + }], + })}\n\n`); + frames.push(`data: ${JSON.stringify({ choices: [{ index: 0, delta: {}, finish_reason: "stop" }] })}\n\n`); + frames.push("data: [DONE]\n\n"); + return frames; +} + const upstream = Bun.serve({ hostname: "127.0.0.1", port: 0, @@ -248,6 +281,9 @@ const upstream = Bun.serve({ }, }), { headers: { "content-type": "text/event-stream" } }); } + return new Response(streamFrames(buildCancelFrames(sessionId), stableHash(sessionId, options.seed), 10), { + headers: { "content-type": "text/event-stream" }, + }); } const frames = buildToolFrames(body); return new Response(streamFrames(frames, stableHash(sessionId, options.seed)), { @@ -256,16 +292,14 @@ const upstream = Bun.serve({ }, }); -let childStdoutTail = ""; -let childStderrTail = ""; const child = Bun.spawn({ cmd: [ process.execPath, - `${import.meta.dir}/memory-recall-soak-child.ts`, + join(import.meta.dir, "memory-recall-soak-child.ts"), "--upstream", upstream.url.toString().replace(/\/$/, ""), ], - cwd: `${import.meta.dir}/..`, + cwd: join(import.meta.dir, ".."), stdout: "pipe", stderr: "pipe", }); @@ -284,9 +318,8 @@ async function consumeChildStdout(): Promise { while (true) { const { done, value } = await reader.read(); if (done) break; - const text = decoder.decode(value, { stream: true }); - childStdoutTail = boundedTail(childStdoutTail, text); - pending += text; + pending += decoder.decode(value, { stream: true }); + if (pending.length > 16_384) pending = pending.slice(-16_384); for (;;) { const newline = pending.indexOf("\n"); if (newline < 0) break; @@ -300,23 +333,18 @@ async function consumeChildStdout(): Promise { resolveReady = null; rejectReady = null; } - } catch { /* bounded tail remains available on failure */ } + } catch { /* child runtime logs are deliberately discarded */ } } } } -async function consumeChildStderr(): Promise { +async function discardChildStderr(): Promise { const reader = child.stderr.getReader(); - const decoder = new TextDecoder(); - while (true) { - const { done, value } = await reader.read(); - if (done) break; - childStderrTail = boundedTail(childStderrTail, decoder.decode(value, { stream: true })); - } + while (!(await reader.read()).done) { /* drain without retaining local paths or payloads */ } } void consumeChildStdout().catch(error => rejectReady?.(error instanceof Error ? error : new Error(String(error)))); -void consumeChildStderr(); +void discardChildStderr(); void child.exited.then(code => { if (resolveReady) rejectReady?.(new Error(`proxy child exited before readiness with code ${code}`)); }); @@ -335,32 +363,62 @@ async function withTimeout(promise: Promise, ms: number, label: string): P } } -const ready = await withTimeout(readyPromise, 30_000, "proxy child readiness"); -const proxyBase = new URL(ready.proxyUrl); -const controlBase = new URL(ready.controlUrl); - -async function sampleMetrics(): Promise { - const response = await fetch(new URL("/metrics", controlBase), { cache: "no-store" }); - if (!response.ok) throw new Error(`metrics endpoint returned ${response.status}`); - return await response.json() as ProbeMetrics; +function observedCurrentBytes(metrics: ProbeMetrics): number { + return Object.values(metrics.appOwnedBytes.observedInFlight) + .reduce((sum, row) => sum + row.currentBytes, 0); } -function idleInvariant(metrics: ProbeMetrics): boolean { - if (metrics.activeTurnCount !== 0) return false; - return Object.values(metrics.appOwnedBytes.observedInFlight).every(row => row.currentBytes === 0 && row.active === 0); +function observedHighWaterBytes(metrics: ProbeMetrics): number { + return Object.values(metrics.appOwnedBytes.observedInFlight) + .reduce((sum, row) => sum + row.highWaterBytes, 0); } -async function waitForIdle(): Promise { - const deadline = Date.now() + options.idleDeadlineMs; - let latest = await sampleMetrics(); - while (!idleInvariant(latest) && Date.now() < deadline) { - await Bun.sleep(Math.min(100, options.sampleIntervalMs)); - latest = await sampleMetrics(); - } - if (!idleInvariant(latest)) { - throw new Error(`proxy did not return to app-owned idle invariants within ${options.idleDeadlineMs} ms`); +function metricPeaks(samples: readonly ProbeMetrics[]): MetricPeaks { + if (samples.length === 0) throw new Error("wave collected no memory samples"); + const peaks: MetricPeaks = { + rss: 0, + heapUsed: 0, + heapTotal: 0, + external: 0, + arrayBuffers: 0, + activeTurnCount: 0, + retainedBytes: 0, + observedInFlightBytes: 0, + observedInFlightHighWaterBytes: 0, + }; + for (const sample of samples) { + peaks.rss = Math.max(peaks.rss, sample.rss); + peaks.heapUsed = Math.max(peaks.heapUsed, sample.heapUsed); + peaks.heapTotal = Math.max(peaks.heapTotal, sample.heapTotal); + peaks.external = Math.max(peaks.external, sample.external); + peaks.arrayBuffers = Math.max(peaks.arrayBuffers, sample.arrayBuffers); + peaks.activeTurnCount = Math.max(peaks.activeTurnCount, sample.activeTurnCount); + peaks.retainedBytes = Math.max(peaks.retainedBytes, sample.appOwnedBytes.retainedBytes); + peaks.observedInFlightBytes = Math.max(peaks.observedInFlightBytes, observedCurrentBytes(sample)); + peaks.observedInFlightHighWaterBytes = Math.max( + peaks.observedInFlightHighWaterBytes, + observedHighWaterBytes(sample), + ); } - return latest; + return peaks; +} + +function mergePeaks(peaks: readonly MetricPeaks[]): MetricPeaks { + if (peaks.length === 0) throw new Error("no wave peaks recorded"); + return peaks.reduce((merged, current) => ({ + rss: Math.max(merged.rss, current.rss), + heapUsed: Math.max(merged.heapUsed, current.heapUsed), + heapTotal: Math.max(merged.heapTotal, current.heapTotal), + external: Math.max(merged.external, current.external), + arrayBuffers: Math.max(merged.arrayBuffers, current.arrayBuffers), + activeTurnCount: Math.max(merged.activeTurnCount, current.activeTurnCount), + retainedBytes: Math.max(merged.retainedBytes, current.retainedBytes), + observedInFlightBytes: Math.max(merged.observedInFlightBytes, current.observedInFlightBytes), + observedInFlightHighWaterBytes: Math.max( + merged.observedInFlightHighWaterBytes, + current.observedInFlightHighWaterBytes, + ), + })); } function tools(): Array> { @@ -477,89 +535,140 @@ function appendRecallOutput(input: Array>, item: Record< input.push({ type: "function_call_output", call_id: callId, output: marker }); } -async function runSession(sessionId: string, rounds: number, slow: boolean): Promise { - const input: Array> = [{ - type: "message", - role: "user", - content: [{ type: "input_text", text: `${sessionMarker(sessionId)} run the synthetic tools` }], - }]; - let toolCalls = 0; +function makeSessionState(name: string, index: number): SessionState { + const id = `${name}-${index}`; + return { + id, + input: [{ + type: "message", + role: "user", + content: [{ type: "input_text", text: `${sessionMarker(id)} run the synthetic tools` }], + }], + slow: deterministicPercent(id, "slow", options.seed) < options.slowConsumerPercent, + requests: 0, + toolCalls: 0, + }; +} - for (let round = 0; round < rounds; round++) { - const response = await fetch(new URL("/v1/responses", proxyBase), { - method: "POST", - headers: { "content-type": "application/json" }, - body: JSON.stringify({ - model: "mock/test-model", - stream: true, - store: false, - input, - tools: tools(), - tool_choice: "auto", - }), - }); - if (response.status !== 200) { - const detail = (await response.text()).slice(0, 240); - throw new Error(`round ${round} returned HTTP ${response.status}: ${detail}`); - } - const text = await readResponseText(response, slow); - const completed = completedResponseFromSse(text); - if (!completed || completed.status !== "completed" || !Array.isArray(completed.output)) { - throw new Error(`round ${round} did not produce response.completed`); - } - const calls = callableItems(completed.output); - const expected = deterministicToolCount(sessionId, round, options.seed); - if (calls.length !== expected) { - throw new Error(`round ${round} completed ${calls.length} tool calls; expected ${expected}`); - } - const callIds = new Set(calls.map(item => item.call_id)); - if (callIds.size !== calls.length || callIds.has(undefined)) { - throw new Error(`round ${round} produced missing or duplicate call ids`); - } - toolCalls += calls.length; - for (const item of calls) appendRecallOutput(input, item, round); +async function runSessionRound(state: SessionState, round: number, proxyBase: URL): Promise { + state.requests += 1; + const response = await fetch(new URL("/v1/responses", proxyBase), { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model: "mock/test-model", + stream: true, + store: false, + input: state.input, + tools: tools(), + tool_choice: "auto", + }), + }); + if (response.status !== 200) { + const detail = (await response.text()).slice(0, 240); + throw new Error(`round ${round} returned HTTP ${response.status}: ${detail}`); + } + const text = await readResponseText(response, state.slow); + const completed = completedResponseFromSse(text); + if (!completed || completed.status !== "completed" || !Array.isArray(completed.output)) { + throw new Error(`round ${round} did not produce response.completed`); } - return { requests: rounds, toolCalls }; + const calls = callableItems(completed.output); + const expected = deterministicToolCount(state.id, round, options.seed); + if (calls.length !== expected) { + throw new Error(`round ${round} completed ${calls.length} tool calls; expected ${expected}`); + } + const callIds = new Set(calls.map(item => item.call_id)); + if (callIds.size !== calls.length || callIds.has(undefined)) { + throw new Error(`round ${round} produced missing or duplicate call ids`); + } + state.toolCalls += calls.length; + for (const item of calls) appendRecallOutput(state.input, item, round); + + // Tool execution is external to OpenCodex. Model 1..N parallel tool calls by + // delaying the recall by the slowest deterministic synthetic tool in this round. + const parallelToolLatencyMs = calls.reduce((maximum, _item, index) => Math.max( + maximum, + 10 + (stableHash(`tool-latency:${state.id}:${round}:${index}`, options.seed) % 491), + ), 0); + if (parallelToolLatencyMs > 0) await Bun.sleep(parallelToolLatencyMs); } -function peakMetrics(samples: readonly ProbeMetrics[]): ProbeMetrics { - if (samples.length === 0) throw new Error("wave collected no memory samples"); - return samples.reduce((peak, sample) => sample.rss > peak.rss ? sample : peak); +async function sampleMetrics(controlBase: URL): Promise { + const response = await fetch(new URL("/metrics", controlBase), { cache: "no-store" }); + if (!response.ok) throw new Error(`metrics endpoint returned ${response.status}`); + return await response.json() as ProbeMetrics; } -async function runWave(name: string, sessions: number, rounds: number): Promise { +function idleInvariant(metrics: ProbeMetrics): boolean { + if (metrics.activeTurnCount !== 0) return false; + return Object.values(metrics.appOwnedBytes.observedInFlight) + .every(row => row.currentBytes === 0 && row.active === 0); +} + +async function waitForIdle(controlBase: URL): Promise { + const deadline = Date.now() + options.idleDeadlineMs; + let latest = await sampleMetrics(controlBase); + while (!idleInvariant(latest) && Date.now() < deadline) { + await Bun.sleep(Math.min(100, options.sampleIntervalMs)); + latest = await sampleMetrics(controlBase); + } + if (!idleInvariant(latest)) { + throw new Error(`proxy did not return to app-owned idle invariants within ${options.idleDeadlineMs} ms`); + } + return latest; +} + +async function runWave( + name: string, + sessions: number, + rounds: number, + proxyBase: URL, + controlBase: URL, +): Promise { const started = Date.now(); + const states = Array.from({ length: sessions }, (_, index) => makeSessionState(name, index)); const samples: ProbeMetrics[] = []; let monitoring = true; const monitor = (async () => { while (monitoring) { - try { samples.push(await sampleMetrics()); } catch { /* main wave outcome remains authoritative */ } + try { samples.push(await sampleMetrics(controlBase)); } catch { /* wave outcome remains authoritative */ } if (monitoring) await Bun.sleep(options.sampleIntervalMs); } })(); - const settled = await Promise.allSettled(Array.from({ length: sessions }, (_, index) => { - const sessionId = `${name}-${index}`; - const slow = deterministicPercent(sessionId, "slow", options.seed) < options.slowConsumerPercent; - return runSession(sessionId, rounds, slow); - })); + // The outer round loop is the barrier: every still-healthy session completes + // its model stream plus external-tool latency before the next recall wave starts. + for (let round = 0; round < rounds; round++) { + const active = states.filter(state => state.failure === undefined); + if (active.length === 0) break; + const settled = await Promise.allSettled(active.map(state => runSessionRound(state, round, proxyBase))); + for (let index = 0; index < settled.length; index++) { + const outcome = settled[index]; + if (outcome.status === "rejected") { + active[index].failure = String( + outcome.reason instanceof Error ? outcome.reason.message : outcome.reason, + ).slice(0, 240); + } + } + } + monitoring = false; await monitor; - samples.push(await sampleMetrics()); - - const failures = settled.filter(result => result.status === "rejected"); - const successes = settled.filter((result): result is PromiseFulfilledResult => result.status === "fulfilled"); - const idle = await waitForIdle(); + samples.push(await sampleMetrics(controlBase)); + const idle = await waitForIdle(controlBase); samples.push(idle); + const completedSessions = states.filter(state => state.failure === undefined && state.requests === rounds).length; + const failedSessions = states.length - completedSessions; const result: WaveResult = { name, sessions, rounds, - completedSessions: successes.length, - failedSessions: failures.length, - requestCount: successes.reduce((sum, result) => sum + result.value.requests, 0), - toolCallCount: successes.reduce((sum, result) => sum + result.value.toolCalls, 0), - peak: peakMetrics(samples), + completedSessions, + failedSessions, + requestCount: states.reduce((sum, state) => sum + state.requests, 0), + toolCallCount: states.reduce((sum, state) => sum + state.toolCalls, 0), + peaks: metricPeaks(samples), idle, durationMs: Date.now() - started, }; @@ -568,22 +677,20 @@ async function runWave(name: string, sessions: number, rounds: number): Promise< name, sessions, rounds, - completedSessions: result.completedSessions, - failedSessions: result.failedSessions, + completedSessions, + failedSessions, requestCount: result.requestCount, toolCallCount: result.toolCallCount, - peakRss: result.peak.rss, + peaks: result.peaks, idleRss: idle.rss, idleRetainedBytes: idle.appOwnedBytes.retainedBytes, durationMs: result.durationMs, - firstFailure: failures[0]?.status === "rejected" - ? String(failures[0].reason instanceof Error ? failures[0].reason.message : failures[0].reason).slice(0, 240) - : undefined, + firstFailure: states.find(state => state.failure)?.failure, }); return result; } -async function cancelOneResponse(sessionId: string): Promise { +async function cancelOneResponse(sessionId: string, proxyBase: URL): Promise { const controller = new AbortController(); const response = await fetch(new URL("/v1/responses", proxyBase), { method: "POST", @@ -609,10 +716,10 @@ async function cancelOneResponse(sessionId: string): Promise { return "cancelled"; } -async function runFaultSession(index: number): Promise<{ kind: FaultKind; outcome: string }> { +async function runFaultSession(index: number, proxyBase: URL): Promise<{ kind: FaultKind; outcome: string }> { const sessionId = `fault-${index}`; const kind = faultKind(sessionId); - if (kind === "cancel") return { kind, outcome: await cancelOneResponse(sessionId) }; + if (kind === "cancel") return { kind, outcome: await cancelOneResponse(sessionId, proxyBase) }; try { const response = await fetch(new URL("/v1/responses", proxyBase), { @@ -635,22 +742,29 @@ async function runFaultSession(index: number): Promise<{ kind: FaultKind; outcom } } -async function runFaultWave(): Promise> { +async function runFaultWave(proxyBase: URL, controlBase: URL): Promise> { if (options.faultSessions === 0) return {}; - const results = await Promise.all(Array.from({ length: options.faultSessions }, (_, index) => runFaultSession(index))); + const results = await Promise.all( + Array.from({ length: options.faultSessions }, (_, index) => runFaultSession(index, proxyBase)), + ); const counts: Record = {}; for (const result of results) { const key = `${result.kind}:${result.outcome}`; counts[key] = (counts[key] ?? 0) + 1; } - await waitForIdle(); + await waitForIdle(controlBase); emit({ type: "FAULT_WAVE", sessions: options.faultSessions, outcomes: counts }); return counts; } let exitCode = 0; +let ready: ChildReady | null = null; +let controlBase: URL | null = null; try { - const initial = await waitForIdle(); + ready = await withTimeout(readyPromise, 30_000, "proxy child readiness"); + const proxyBase = new URL(ready.proxyUrl); + controlBase = new URL(ready.controlUrl); + const initial = await waitForIdle(controlBase); emit({ type: "START", seed: options.seed, @@ -663,17 +777,23 @@ try { const sustained: WaveResult[] = []; for (let wave = 0; wave < options.sustainedWaves; wave++) { - sustained.push(await runWave(`sustained-${wave}`, options.sustainedSessions, options.sustainedRounds)); + sustained.push(await runWave( + `sustained-${wave}`, + options.sustainedSessions, + options.sustainedRounds, + proxyBase, + controlBase, + )); } - const burst = await runWave("burst", options.burstSessions, options.burstRounds); - const faultOutcomes = await runFaultWave(); - const final = await waitForIdle(); + const burst = await runWave("burst", options.burstSessions, options.burstRounds, proxyBase, controlBase); + const faultOutcomes = await runFaultWave(proxyBase, controlBase); + const final = await waitForIdle(controlBase); const failedBaselineSessions = sustained.reduce((sum, wave) => sum + wave.failedSessions, 0) + burst.failedSessions; const idleRss = sustained.map(wave => wave.idle.rss); const idleRetained = sustained.map(wave => wave.idle.appOwnedBytes.retainedBytes); - const peakRss = maxFinite([...sustained.map(wave => wave.peak.rss), burst.peak.rss]); - const summary = { + const peaks = mergePeaks([...sustained.map(wave => wave.peaks), burst.peaks]); + emit({ type: "SUMMARY", outcome: failedBaselineSessions === 0 ? "PASS" : "FAIL", seed: options.seed, @@ -683,7 +803,7 @@ try { bunRevision: ready.bunRevision, sustainedWaves: options.sustainedWaves, failedBaselineSessions, - peakRss, + peaks, initialRss: initial.rss, finalRss: final.rss, rssIdleSlopeBytesPerWave: linearSlope(idleRss), @@ -693,9 +813,8 @@ try { finalOverBudgetBytes: final.appOwnedBytes.overBudgetBytes, finalActiveTurnCount: final.activeTurnCount, faultOutcomes, - note: "RSS slope is profiling evidence only; PASS is based on protocol completion and app-owned cleanup invariants.", - }; - emit(summary); + note: "Process-memory slopes are profiling evidence only; PASS is based on protocol completion and app-owned cleanup invariants.", + }); if (failedBaselineSessions !== 0) exitCode = 1; } catch (error) { exitCode = 1; @@ -705,17 +824,16 @@ try { classification: "probe-error", seed: options.seed, failure: error instanceof Error ? error.message : String(error), - childStdoutTail, - childStderrTail, + childExitCode: child.exitCode, }); } finally { - try { - await fetch(new URL("/shutdown", controlBase), { method: "POST" }); - } catch { /* child may already have exited */ } - upstream.stop(true); + if (controlBase) { + try { await fetch(new URL("/shutdown", controlBase), { method: "POST" }); } catch { /* child may already be gone */ } + } + await upstream.stop(true); const childExit = await Promise.race([child.exited, Bun.sleep(2_000).then(() => null)]); if (childExit === null) { - try { child.kill("SIGKILL"); } catch { /* already exited */ } + try { child.kill(); } catch { /* already exited */ } } } From 616cfc76e149183d0cdd259540bf2e8fe9d2ce45 Mon Sep 17 00:00:00 2001 From: Wibias <37517432+Wibias@users.noreply.github.com> Date: Mon, 17 Aug 2026 09:15:56 +0200 Subject: [PATCH 08/11] test(memory): keep repeated soak waves byte-identical --- scripts/memory-recall-soak.ts | 23 ++++++++++++++++++----- 1 file changed, 18 insertions(+), 5 deletions(-) diff --git a/scripts/memory-recall-soak.ts b/scripts/memory-recall-soak.ts index 1c60962f2e..34ea3e561c 100644 --- a/scripts/memory-recall-soak.ts +++ b/scripts/memory-recall-soak.ts @@ -344,7 +344,7 @@ async function discardChildStderr(): Promise { } void consumeChildStdout().catch(error => rejectReady?.(error instanceof Error ? error : new Error(String(error)))); -void discardChildStderr(); +void discardChildStderr().catch(() => {}); void child.exited.then(code => { if (resolveReady) rejectReady?.(new Error(`proxy child exited before readiness with code ${code}`)); }); @@ -535,8 +535,8 @@ function appendRecallOutput(input: Array>, item: Record< input.push({ type: "function_call_output", call_id: callId, output: marker }); } -function makeSessionState(name: string, index: number): SessionState { - const id = `${name}-${index}`; +function makeSessionState(workloadKey: string, index: number): SessionState { + const id = `${workloadKey}-${index}`; return { id, input: [{ @@ -621,13 +621,14 @@ async function waitForIdle(controlBase: URL): Promise { async function runWave( name: string, + workloadKey: string, sessions: number, rounds: number, proxyBase: URL, controlBase: URL, ): Promise { const started = Date.now(); - const states = Array.from({ length: sessions }, (_, index) => makeSessionState(name, index)); + const states = Array.from({ length: sessions }, (_, index) => makeSessionState(workloadKey, index)); const samples: ProbeMetrics[] = []; let monitoring = true; const monitor = (async () => { @@ -684,6 +685,8 @@ async function runWave( peaks: result.peaks, idleRss: idle.rss, idleRetainedBytes: idle.appOwnedBytes.retainedBytes, + idleInspectionCounters: idle.inspectionCounters, + idleResponseState: idle.responseState, durationMs: result.durationMs, firstFailure: states.find(state => state.failure)?.failure, }); @@ -779,13 +782,21 @@ try { for (let wave = 0; wave < options.sustainedWaves; wave++) { sustained.push(await runWave( `sustained-${wave}`, + "sustained", options.sustainedSessions, options.sustainedRounds, proxyBase, controlBase, )); } - const burst = await runWave("burst", options.burstSessions, options.burstRounds, proxyBase, controlBase); + const burst = await runWave( + "burst", + "burst", + options.burstSessions, + options.burstRounds, + proxyBase, + controlBase, + ); const faultOutcomes = await runFaultWave(proxyBase, controlBase); const final = await waitForIdle(controlBase); @@ -812,6 +823,8 @@ try { finalPinnedBytes: final.appOwnedBytes.pinnedBytes, finalOverBudgetBytes: final.appOwnedBytes.overBudgetBytes, finalActiveTurnCount: final.activeTurnCount, + finalInspectionCounters: final.inspectionCounters, + finalResponseState: final.responseState, faultOutcomes, note: "Process-memory slopes are profiling evidence only; PASS is based on protocol completion and app-owned cleanup invariants.", }); From 8bcbcc970e14dd9bd4b6ea2af19d1fde3f15a3e9 Mon Sep 17 00:00:00 2001 From: Wibias <37517432+Wibias@users.noreply.github.com> Date: Mon, 17 Aug 2026 09:21:59 +0200 Subject: [PATCH 09/11] fix(memory): harden soak child startup cleanup --- scripts/memory-recall-soak-child.ts | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/scripts/memory-recall-soak-child.ts b/scripts/memory-recall-soak-child.ts index e800fca86b..3ee0b7a025 100644 --- a/scripts/memory-recall-soak-child.ts +++ b/scripts/memory-recall-soak-child.ts @@ -30,13 +30,22 @@ try { console.error("memory-recall-soak-child: --upstream must be a URL"); process.exit(2); } -if (upstreamUrl.hostname !== "127.0.0.1" && upstreamUrl.hostname !== "localhost" && upstreamUrl.hostname !== "::1") { +const LOOPBACK_HOSTNAMES = new Set(["127.0.0.1", "localhost", "[::1]"]); +if (upstreamUrl.protocol !== "http:" && upstreamUrl.protocol !== "https:") { + console.error("memory-recall-soak-child: --upstream must be http or https"); + process.exit(2); +} +if (!LOOPBACK_HOSTNAMES.has(upstreamUrl.hostname)) { console.error("memory-recall-soak-child: upstream must be loopback"); process.exit(2); } const home = mkdtempSync(join(tmpdir(), "ocx-memory-recall-soak-")); process.env.OPENCODEX_HOME = home; +function cleanupHome(): void { + try { rmSync(home, { recursive: true, force: true }); } catch { /* temp cleanup only */ } +} +process.on("exit", cleanupHome); const [configModule, serverModule, memoryModule, lifecycleModule, relayModule, responseStateModule] = await Promise.all([ import("../src/config"), @@ -94,7 +103,7 @@ async function closeAndExit(code: number): Promise { closing = true; try { await proxy.stop(true); } catch { /* best-effort probe teardown */ } try { control?.stop(true); } catch { /* best-effort probe teardown */ } - try { rmSync(home, { recursive: true, force: true }); } catch { /* temp cleanup only */ } + cleanupHome(); process.exit(code); } From f6bd75388ffd5e2300ce2074a278d75c26bd6a6e Mon Sep 17 00:00:00 2001 From: Wibias <37517432+Wibias@users.noreply.github.com> Date: Mon, 17 Aug 2026 09:22:35 +0200 Subject: [PATCH 10/11] test(memory): cover soak option and rng bounds --- tests/memory-recall-soak.test.ts | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/tests/memory-recall-soak.test.ts b/tests/memory-recall-soak.test.ts index f354ca2db2..89496b8fc7 100644 --- a/tests/memory-recall-soak.test.ts +++ b/tests/memory-recall-soak.test.ts @@ -44,6 +44,8 @@ describe("#820 memory recall soak probe helpers", () => { expect(() => parseMemoryRecallSoakOptions(["--slow-percent", "101"])).toThrow(); expect(() => parseMemoryRecallSoakOptions(["--unknown"])).toThrow(); expect(() => parseMemoryRecallSoakOptions(["positional"])).toThrow(); + expect(() => parseMemoryRecallSoakOptions(["--sessions"])).toThrow(); + expect(() => parseMemoryRecallSoakOptions(["--sessions", "--rounds", "2"])).toThrow(); }); test("seeded workload decisions are reproducible and remain in bounds", () => { @@ -51,6 +53,13 @@ describe("#820 memory recall soak probe helpers", () => { const second = mulberry32(820_001); expect(Array.from({ length: 8 }, () => first())).toEqual(Array.from({ length: 8 }, () => second())); + const ranged = mulberry32(1); + for (let index = 0; index < 64; index++) { + const value = ranged(); + expect(value).toBeGreaterThanOrEqual(0); + expect(value).toBeLessThan(1); + } + for (let index = 0; index < 128; index++) { const session = `session-${index}`; expect(stableHash(session, 7)).toBe(stableHash(session, 7)); From bc4baf31b32cda2b1481c8fea933b7d03b38f593 Mon Sep 17 00:00:00 2001 From: Wibias <37517432+Wibias@users.noreply.github.com> Date: Mon, 17 Aug 2026 09:23:51 +0200 Subject: [PATCH 11/11] fix(memory): bound soak probe requests --- scripts/memory-recall-soak.ts | 64 ++++++++++++++++++++++------------- 1 file changed, 41 insertions(+), 23 deletions(-) diff --git a/scripts/memory-recall-soak.ts b/scripts/memory-recall-soak.ts index 34ea3e561c..e3f894c936 100644 --- a/scripts/memory-recall-soak.ts +++ b/scripts/memory-recall-soak.ts @@ -153,6 +153,8 @@ function advertisedToolNames(body: Record): string[] { } function orderedToolNames(names: readonly string[]): string[] { + // extractRound advances from markers written by function/custom outputs. Keep a + // marker-producing tool first so every non-empty 1..N selection advances the round. const ordinary = names.filter(name => !/apply_patch|tool_search/i.test(name)); const special = names.filter(name => /apply_patch|tool_search/i.test(name)); return [...ordinary, ...special]; @@ -475,14 +477,13 @@ function toolSearchResultTools(): Array> { }]; } -async function readResponseText(response: Response, slow: boolean, signal?: AbortSignal): Promise { +async function readResponseText(response: Response, slow: boolean): Promise { if (!response.body) return ""; const reader = response.body.getReader(); const decoder = new TextDecoder(); let text = ""; try { while (true) { - if (signal?.aborted) throw signal.reason ?? new Error("aborted"); const { done, value } = await reader.read(); if (done) break; text += decoder.decode(value, { stream: true }); @@ -554,6 +555,7 @@ async function runSessionRound(state: SessionState, round: number, proxyBase: UR state.requests += 1; const response = await fetch(new URL("/v1/responses", proxyBase), { method: "POST", + signal: AbortSignal.timeout(options.idleDeadlineMs), headers: { "content-type": "application/json" }, body: JSON.stringify({ model: "mock/test-model", @@ -595,7 +597,10 @@ async function runSessionRound(state: SessionState, round: number, proxyBase: UR } async function sampleMetrics(controlBase: URL): Promise { - const response = await fetch(new URL("/metrics", controlBase), { cache: "no-store" }); + const response = await fetch(new URL("/metrics", controlBase), { + cache: "no-store", + signal: AbortSignal.timeout(Math.min(5_000, options.idleDeadlineMs)), + }); if (!response.ok) throw new Error(`metrics endpoint returned ${response.status}`); return await response.json() as ProbeMetrics; } @@ -695,28 +700,35 @@ async function runWave( async function cancelOneResponse(sessionId: string, proxyBase: URL): Promise { const controller = new AbortController(); - const response = await fetch(new URL("/v1/responses", proxyBase), { - method: "POST", - signal: controller.signal, - headers: { "content-type": "application/json" }, - body: JSON.stringify({ - model: "mock/test-model", - stream: true, - store: false, - input: sessionMarker(sessionId), - tools: tools(), - }), - }); - const reader = response.body?.getReader(); - if (!reader) throw new Error("cancel probe response had no body"); + const timeout = setTimeout(() => { + controller.abort(new Error(`cancel probe timed out after ${options.idleDeadlineMs} ms`)); + }, options.idleDeadlineMs); try { - await reader.read(); - controller.abort(new Error("synthetic client cancel")); - try { await reader.read(); } catch { /* cancellation is the expected outcome */ } + const response = await fetch(new URL("/v1/responses", proxyBase), { + method: "POST", + signal: controller.signal, + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model: "mock/test-model", + stream: true, + store: false, + input: sessionMarker(sessionId), + tools: tools(), + }), + }); + const reader = response.body?.getReader(); + if (!reader) throw new Error("cancel probe response had no body"); + try { + await reader.read(); + controller.abort(new Error("synthetic client cancel")); + try { await reader.read(); } catch { /* cancellation is the expected outcome */ } + } finally { + try { await reader.cancel(); } catch { /* already aborted */ } + } + return "cancelled"; } finally { - try { await reader.cancel(); } catch { /* already aborted */ } + clearTimeout(timeout); } - return "cancelled"; } async function runFaultSession(index: number, proxyBase: URL): Promise<{ kind: FaultKind; outcome: string }> { @@ -727,6 +739,7 @@ async function runFaultSession(index: number, proxyBase: URL): Promise<{ kind: F try { const response = await fetch(new URL("/v1/responses", proxyBase), { method: "POST", + signal: AbortSignal.timeout(options.idleDeadlineMs), headers: { "content-type": "application/json" }, body: JSON.stringify({ model: "mock/test-model", @@ -841,7 +854,12 @@ try { }); } finally { if (controlBase) { - try { await fetch(new URL("/shutdown", controlBase), { method: "POST" }); } catch { /* child may already be gone */ } + try { + await fetch(new URL("/shutdown", controlBase), { + method: "POST", + signal: AbortSignal.timeout(Math.min(2_000, options.idleDeadlineMs)), + }); + } catch { /* child may already be gone */ } } await upstream.stop(true); const childExit = await Promise.race([child.exited, Bun.sleep(2_000).then(() => null)]);