diff --git a/examples/coordination/sqlite-capacity-report.ts b/examples/coordination/sqlite-capacity-report.ts index b586d2f292..8580fff821 100644 --- a/examples/coordination/sqlite-capacity-report.ts +++ b/examples/coordination/sqlite-capacity-report.ts @@ -17,6 +17,44 @@ export interface CapacityScanInput { skipped_private_files: number; } +/** + * Split a reader command emitted by the quota context-capture navigation the + * same way its generator quotes it. The producer builds commands with POSIX + * `shlex.join`, so a quoted path — including a temporary directory containing + * spaces or an apostrophe — is exactly one argument, never split on its own + * whitespace. Unbalanced quoting fails loudly instead of mis-parsing a + * reference path into several arguments. + */ +export function splitShellCommand(command: string): string[] { + const argv: string[] = []; + let current = ""; + let started = false; + let quote: "'" | "\"" | null = null; + const characters = Array.from(command); + for (let index = 0; index < characters.length; index++) { + const character = characters[index]!; + if (quote === "'") { + if (character === "'") quote = null; + else current += character; + } else if (quote === "\"") { + if (character === "\"") quote = null; + else if (character === "\\" && index + 1 < characters.length && + "\"\\$`".includes(characters[index + 1]!)) current += characters[++index]!; + else current += character; + } else if (character === "'" || character === "\"") { + quote = character; started = true; + } else if (character === " " || character === "\t") { + if (started || current !== "") argv.push(current); + current = ""; started = false; + } else { + current += character; started = true; + } + } + if (quote !== null) throw new Error(`unbalanced ${quote} in reader command`); + if (started || current !== "") argv.push(current); + return argv; +} + export function sameCapacityScan(left: CapacityScanInput | undefined, right: CapacityScanInput | undefined): boolean { return !!left && !!right && /^[a-f0-9]{64}$/.test(left.sha256) && Number.isSafeInteger(left.files) && left.files > 0 && diff --git a/examples/coordination/sqlite-capacity.ts b/examples/coordination/sqlite-capacity.ts index 6e8fcf2d22..9174302342 100644 --- a/examples/coordination/sqlite-capacity.ts +++ b/examples/coordination/sqlite-capacity.ts @@ -17,8 +17,8 @@ import {selectLocalSqliteAuthority} from "../../loopx/control_plane/coordination import {engageLegacyCoordinationWriterFence} from "../../loopx/control_plane/coordination/legacy_writer_fence.ts"; import {canonicalAuthoritySha256} from "../../loopx/control_plane/coordination/authority_store_codec.ts"; import {authorityProjectionFixture} from "../../tests/control_plane_ts/authority_projection_fixture.ts"; -import {CAPACITY_PROFILES, capacityLedger, latency, type CapacityAxis, type CapacityProfileId, - type QualificationRow} from "./sqlite-capacity-report.ts"; +import {CAPACITY_PROFILES, capacityLedger, latency, splitShellCommand, + type CapacityAxis, type CapacityProfileId, type QualificationRow} from "./sqlite-capacity-report.ts"; const {values: options} = parseArgs({options: { profile: {type: "string", default: "rehearsal"}, output: {type: "string"}, @@ -148,23 +148,54 @@ async function measureAxis(count: number): Promise { let phase = "setup", managedRuntimeUsed = false; const stopRuntime = () => { if (!managedRuntimeUsed) return; + // Cleanup deletes the runtime's whole temporary namespace, so the stop is + // anchored to observed process exit: a shutdown receipt alone let the + // runtime's exit-time lock retirement mkdir the axis directory back after + // cleanup_verified was already recorded. Reuse the disposable-owner stop + // interface and require its result to prove the exit before any delete. const child = spawnSync(options.python!, ["-c", - "from loopx.control_plane.effect_runtime import effect_runtime_result; effect_runtime_result('runtime.shutdown', {}, retry_safe=False)"], + "import json; from loopx.control_plane.effect_runtime import restart_effect_runtime; " + + "print(json.dumps(restart_effect_runtime(wait_for_exit=True)))"], {cwd: repository, env: environment, encoding: "utf8", timeout: 30000}); assert.equal(child.status, 0, "isolated Effect runtime shutdown failed"); + assert.equal((JSON.parse(child.stdout) as Record).stopped, true, + "isolated Effect runtime did not exit before cleanup"); managedRuntimeUsed = false; }; const cli = (args: string[], samples: number[]) => { phase = `cold_cli_${args[0] === "todo" ? "mutation" : args[0]}`; const start = performance.now(); managedRuntimeUsed = true; - const child = spawnSync(options.python!, ["-m", "loopx.cli", "--registry", registry, "--format", "json", ...args], + const run = (file: string, argv: string[]) => spawnSync(file, argv, {cwd: repository, env: environment, encoding: "utf8", timeout: 60000, maxBuffer: 4 * 1024 ** 2}); - const elapsed = performance.now() - start; + const child = run(options.python!, ["-m", "loopx.cli", "--registry", registry, "--format", "json", ...args]); if (child.status !== 0) process.stderr.write(`${phase}: process exited ${child.status ?? "without an exit code"}\n`); assert.equal(child.status, 0, `cold CLI ${args[0]} failed`); const result = JSON.parse(child.stdout) as Record; assert.notEqual(result.ok, false, `cold CLI ${args[0]} rejected`); - samples.push(elapsed); stopRuntime(); return result; + if (result.schema_version === "context_capture_navigation_v1") { + // `quota should-run` delivers its payload through a context capture: + // follow the returned reader commands page by page until eof, then parse + // the reassembled original JSON. The sample keeps the whole navigation + // cost, because a cold reader cannot act before the payload arrives. + let follow = splitShellCommand(String(result.command ?? result.first_read_command ?? "")); + assert.ok(follow.length > 0, `cold CLI ${args[0]} navigation returned no read command`); + let payload = ""; + for (let pages = 0; ; pages++) { + assert.ok(pages < 256, `cold CLI ${args[0]} context read exceeded the page guard`); + const read = run(follow[0]!, follow.slice(1)); + if (read.status !== 0) process.stderr.write(`${phase}: context read exited ${read.status ?? "without an exit code"}\n`); + assert.equal(read.status, 0, `cold CLI ${args[0]} context read failed`); + const page = JSON.parse(read.stdout) as Record; + assert.equal(page.ok, true, `cold CLI ${args[0]} context read rejected`); + payload += typeof page.content === "string" ? page.content : ""; + if (page.eof === true) break; + const next = page.next_command; + assert.ok(typeof next === "string" && next.length > 0, `cold CLI ${args[0]} context page lacks the next command before eof`); + follow = splitShellCommand(next); + } + samples.push(performance.now() - start); stopRuntime(); return JSON.parse(payload) as Record; + } + samples.push(performance.now() - start); stopRuntime(); return result; }; try { assert.equal((await selectLocalSqliteAuthority(runtime, goal, true)).ok, true); @@ -382,12 +413,26 @@ async function measureAxis(count: number): Promise { } axis.status = "passed"; } catch (error) { - axis.failure = `${phase}: ${error instanceof Error ? error.name : "unknown failure"}`; + axis.failure = `${phase}: ${error instanceof Error ? `${error.name}: ${error.message.slice(0, 600)}` : "unknown failure"}`; } finally { - try { stopRuntime(); } catch { axis.status = "failed"; axis.failure = "isolated_runtime_cleanup_failed"; } + // A stop that cannot prove observed process exit fails the axis, but the + // first recorded failure (the measurement error) keeps precedence. + try { stopRuntime(); } + catch { axis.status = "failed"; if (!axis.failure) axis.failure = "isolated_runtime_cleanup_failed"; } axis.resource_peak_rss_bytes = process.resourceUsage().maxRSS * 1024; - try { rmSync(root, {recursive: true, force: true}); axis.cleanup_verified = !existsSync(root); } - catch { axis.status = "failed"; axis.failure = "temporary_database_cleanup_failed"; } + // A just-stopped managed runtime can leave a grandchild briefly holding the + // temp root (for example its working directory), which makes an immediate + // recursive delete raise while a retry a moment later succeeds. Retry a few + // times before declaring the disposable database uncleaned. + for (let attempt = 0; ; attempt++) { + try { rmSync(root, {recursive: true, force: true}); } catch { /* retried below */ } + axis.cleanup_verified = !existsSync(root); + if (axis.cleanup_verified || attempt >= 4) break; + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 250); + } + if (!axis.cleanup_verified && axis.status !== "failed") { + axis.status = "failed"; axis.failure = "temporary_database_cleanup_failed"; + } } return axis; } diff --git a/tests/control_plane_ts/sqlite_capacity.test.ts b/tests/control_plane_ts/sqlite_capacity.test.ts index a0171d6bfe..dd2401e36a 100644 --- a/tests/control_plane_ts/sqlite_capacity.test.ts +++ b/tests/control_plane_ts/sqlite_capacity.test.ts @@ -6,13 +6,47 @@ import {tmpdir} from "node:os"; import {performance} from "node:perf_hooks"; import {fileURLToPath} from "node:url"; import test from "node:test"; -import {capacityLedger, latency, sameCapacityScan, type CapacityAxis} from "../../examples/coordination/sqlite-capacity-report.ts"; +import {capacityLedger, latency, sameCapacityScan, splitShellCommand, + type CapacityAxis} from "../../examples/coordination/sqlite-capacity-report.ts"; test("latency has explicit nearest-rank tails and rejects absent or invalid samples", () => { assert.deepEqual(latency([5, 1, 4, 2, 3]), {n: 5, p50_ms: 3, p95_ms: 5, p99_ms: 5}); for (const samples of [[], [NaN], [Infinity], [-1]]) assert.throws(() => latency(samples)); }); +test("reader commands split exactly as shlex.join quotes them", () => { + assert.deepEqual(splitShellCommand("python3 -m loopx.cli --format json quota context-read --offset 0"), + ["python3", "-m", "loopx.cli", "--format", "json", "quota", "context-read", "--offset", "0"]); + // shlex.join single-quotes a path containing spaces: it stays one argument. + assert.deepEqual(splitShellCommand( + `python3 -m loopx.cli quota context-read --reference-file '${"/tmp/space dir/capacity.reader.json"}' --offset 0`), + ["python3", "-m", "loopx.cli", "quota", "context-read", "--reference-file", + "/tmp/space dir/capacity.reader.json", "--offset", "0"]); + // An apostrophe inside the path uses the POSIX '"'"' escape sequence. + assert.deepEqual(splitShellCommand( + `python3 -m loopx.cli quota context-read --reference-file '/tmp/loop'"'"'x y/capacity.reader.json' --offset 0`), + ["python3", "-m", "loopx.cli", "quota", "context-read", "--reference-file", + "/tmp/loop'x y/capacity.reader.json", "--offset", "0"]); + // Double-quoted arguments and escaped quotes inside them survive too. + assert.deepEqual(splitShellCommand(`python3 "a b" "say \\"hi\\"" c`), ["python3", "a b", 'say "hi"', "c"]); + // A mis-quoted command fails loudly instead of splitting a reference path. + assert.throws(() => splitShellCommand("python3 --reference-file '/tmp/space dir/x")); +}); + +test("cleanup is anchored to observed runtime exit and keeps the first failure", async () => { + // The stop lives in the runner script itself: a shutdown receipt alone let + // the retiring runtime's delayed lock mkdir recreate the axis directories + // after cleanup_verified was recorded. The stop must reuse the disposable + // owner's observed-exit interface, check its result, and never let a stop + // failure overwrite the first recorded measurement error. + const source = await readFile( + fileURLToPath(new URL("../../examples/coordination/sqlite-capacity.ts", import.meta.url)), "utf8"); + assert.match(source, /restart_effect_runtime\(wait_for_exit=True\)/); + assert.doesNotMatch(source, /runtime\.shutdown/); + assert.match(source, /\.stopped, true,\s*\n?\s*"isolated Effect runtime did not exit before cleanup"/); + assert.match(source, /if \(!axis\.failure\) axis\.failure = "isolated_runtime_cleanup_failed"/); +}); + function axis(count: number, payloadBytes = 65536): CapacityAxis { const sample = (n: number) => ({n, p50_ms: 1, p95_ms: 2, p99_ms: 3}); return {target_commits: count, completed_commits: count, projection_json_bytes: payloadBytes,