From f9c7afc32e723923a25e9a8f62ba64ebc26f3113 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 18 Oct 2026 13:13:32 +0800 Subject: [PATCH] fix(checkpoint): bound artifact member IO Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../runtime/checkpoint_artifacts.ts | 122 ++++++++++++------ .../checkpoint_artifacts.test.ts | 83 ++++++++++++ 2 files changed, 163 insertions(+), 42 deletions(-) diff --git a/loopx/control_plane/runtime/checkpoint_artifacts.ts b/loopx/control_plane/runtime/checkpoint_artifacts.ts index 64a3549838..cf66252267 100644 --- a/loopx/control_plane/runtime/checkpoint_artifacts.ts +++ b/loopx/control_plane/runtime/checkpoint_artifacts.ts @@ -21,6 +21,23 @@ const SCHEMA = "loopx_checkpoint_artifact_set_v0"; type Component = {id: string; source_path: string}; type Member = {path: string; kind: "file" | "directory"; mode: number; size?: number; sha256?: string}; +// Keep independent member IO bounded, including on failure: staging cleanup +// must not race copies or flushes that are still running after a rejection. +async function mapMembers(items: T[], visit: (item: T) => Promise): Promise { + const result = new Array(items.length); + let next = 0, failed = false; + let failure: unknown; + await Promise.all(Array.from({length: Math.min(8, items.length)}, async () => { + while (!failed && next < items.length) { + const index = next++; + try {result[index] = await visit(items[index]);} + catch (error) {if (!failed) {failed = true; failure = error;}} + } + })); + if (failed) throw failure; + return result; +} + function absolute(value: unknown, label: string): string { const path = requireNonEmptyString(value, label); if (!isAbsolute(path) || resolve(path) !== path) throw new Error(`${label} must be normalized and absolute`); @@ -61,21 +78,41 @@ async function members(root: string, sourceObservation: boolean): Promise { - const stat = await lstat(path); - if (stat.isSymbolicLink()) throw new Error("checkpoint source contains a symlink"); - // SQLite's reader index and an empty WAL contain no committed data. A - // read-only backup can create either on a closed WAL database. Exclude - // only these recognized sidecars, not a nonempty WAL or an explicitly - // requested standalone file. Artifact verification excludes nothing. - if (sourceObservation && name !== "" && stat.isFile() && - (await sqliteSidecar(path, ["-shm"]) || - (stat.size === 0 && await sqliteSidecar(path, ["-wal"])))) return; - if (stat.isDirectory()) { - output.push({path: name, kind: "directory", mode: stat.mode & 0o777}); - for (const child of (await readdir(path)).sort()) await visit(join(path, child), name ? `${name}/${child}` : child); - } else if (stat.isFile()) { - output.push({path: name, kind: "file", mode: stat.mode & 0o777, size: stat.size, sha256: await digest(path)}); - } else throw new Error("checkpoint source contains an unsupported file type"); + type Node = {member?: Member; children: Node[]}; + const tree: Node = {children: []}; + let frontier = [{path, name, node: tree}]; + while (frontier.length) { + frontier = (await mapMembers(frontier, async entry => { + const {path, name, node} = entry, stat = await lstat(path); + if (stat.isSymbolicLink()) throw new Error("checkpoint source contains a symlink"); + // SQLite's reader index and an empty WAL contain no committed data. A + // read-only backup can create either on a closed WAL database. Exclude + // only these recognized sidecars, not a nonempty WAL or an explicitly + // requested standalone file. Artifact verification excludes nothing. + if (sourceObservation && name !== "" && stat.isFile() && + (await sqliteSidecar(path, ["-shm"]) || + (stat.size === 0 && await sqliteSidecar(path, ["-wal"])))) return []; + if (stat.isDirectory()) { + node.member = {path: name, kind: "directory", mode: stat.mode & 0o777}; + return (await readdir(path)).sort().map(child => { + const childNode: Node = {children: []}; + node.children.push(childNode); + return {path: join(path, child), name: name ? `${name}/${child}` : child, node: childNode}; + }); + } + if (!stat.isFile()) throw new Error("checkpoint source contains an unsupported file type"); + node.member = {path: name, kind: "file", mode: stat.mode & 0o777, size: stat.size, sha256: await digest(path)}; + return []; + })).flat(); + } + // Traversal concurrency cannot change the existing depth-first manifest + // order (which differs from sorting complete relative path strings). + const pending = [tree]; + while (pending.length) { + const node = pending.pop()!; + if (node.member) output.push(node.member); + for (let i = node.children.length - 1; i >= 0; i--) pending.push(node.children[i]); + } } await visit(root, ""); if ((await lstat(root)).isFile() && await sqliteFile(root)) { @@ -137,28 +174,27 @@ async function flushCopiedFile(path: string): Promise { } } async function copyComponent(source: string, destination: string, snapshotDatabases = true): Promise { - const stat = await lstat(source); - if (stat.isSymbolicLink()) throw new Error("checkpoint source became a symlink"); - if (stat.isDirectory()) { - await mkdir(destination, {mode: 0o700}); - const children = (await readdir(source)).sort(); - const databases = new Set(); - for (const child of snapshotDatabases ? children : []) { - const path = join(source, child); - if ((await lstat(path)).isFile()) { - if (await sqliteFile(path)) databases.add(child); + let frontier = [{source, destination, nested: false}]; + while (frontier.length) { + frontier = (await mapMembers(frontier, async entry => { + const {source, destination, nested} = entry, stat = await lstat(source); + if (stat.isSymbolicLink()) throw new Error("checkpoint source became a symlink"); + // An explicitly selected standalone sidecar remains a byte artifact. + // Directory participants seal their databases through the existing WAL + // snapshot owner rather than copying those databases' live sidecars. + if (snapshotDatabases && nested && await sqliteSidecar(source, ["-wal", "-shm", "-journal"])) return []; + if (stat.isDirectory()) { + await mkdir(destination, {mode: 0o700}); + return (await readdir(source)).sort().map(child => ({source: join(source, child), + destination: join(destination, child), nested: true})); } - } - for (const child of children) { - if (["-wal", "-shm", "-journal"].some(suffix => child.endsWith(suffix) && databases.has(child.slice(0, -suffix.length)))) continue; - await copyComponent(join(source, child), join(destination, child), snapshotDatabases); - } - } else if (stat.isFile()) { - // SQLite's existing physical snapshot owner seals committed WAL contents. - if (snapshotDatabases && await sqliteFile(source)) await snapshotSqliteDatabase(source, destination); - else await copyFile(source, destination); - await flushCopiedFile(destination); - } else throw new Error("checkpoint source contains an unsupported file type"); + if (!stat.isFile()) throw new Error("checkpoint source contains an unsupported file type"); + if (snapshotDatabases && await sqliteFile(source)) await snapshotSqliteDatabase(source, destination); + else await copyFile(source, destination); + await flushCopiedFile(destination); + return []; + })).flat(); + } } function manifestBody(raw: unknown): JsonObject { const value = requireJsonObject(raw, "checkpoint manifest"); @@ -357,15 +393,17 @@ export async function checkpointArtifacts(value: unknown): Promise { // that could otherwise escape a before/after comparison. SQLite is sealed // by its existing snapshot owner and has a separate physical witness. for (const c of sources) { - for (const member of before[c.id] as Member[]) { - if (member.kind !== "file") continue; - const source = (await lstat(c.source_path)).isDirectory() ? join(c.source_path, member.path) : c.source_path + member.path; - if (await sqliteFile(source) || (member.path !== "" && await sqliteSidecar(source, ["-wal", "-shm", "-journal"]))) continue; - const copied = (artifacts[c.id] as Member[]).find(m => m.path === member.path && m.kind === "file"); + const directory = (await lstat(c.source_path)).isDirectory(); + const copiedMembers = new Map((artifacts[c.id] as Member[]).filter(m => m.kind === "file").map(m => [m.path, m])); + await mapMembers(before[c.id] as Member[], async member => { + if (member.kind !== "file") return; + const source = directory ? join(c.source_path, member.path) : c.source_path + member.path; + if (await sqliteFile(source) || (member.path !== "" && await sqliteSidecar(source, ["-wal", "-shm", "-journal"]))) return; + const copied = copiedMembers.get(member.path); if (!copied || copied.sha256 !== member.sha256 || copied.size !== member.size || copied.mode !== member.mode) { throw new Error("checkpoint copied bytes do not match the reviewed source"); } - } + }); } const body: JsonObject = {schema_version: SCHEMA, components: sources, point_metadata: requireJsonObject(request.point_metadata, "checkpoint point metadata"), source_witness: before, artifact_witness: artifacts, research_point_qualified: false, diff --git a/tests/control_plane_ts/checkpoint_artifacts.test.ts b/tests/control_plane_ts/checkpoint_artifacts.test.ts index 92f7ceaf00..e35aab5512 100644 --- a/tests/control_plane_ts/checkpoint_artifacts.test.ts +++ b/tests/control_plane_ts/checkpoint_artifacts.test.ts @@ -1,4 +1,7 @@ import assert from "node:assert/strict"; +import {createHash} from "node:crypto"; +import fs from "node:fs/promises"; +import {syncBuiltinESMExports} from "node:module"; import test from "node:test"; import {chmod, mkdir, mkdtemp, readFile, readdir, realpath, rm, stat, symlink, writeFile} from "node:fs/promises"; import {tmpdir} from "node:os"; @@ -56,6 +59,44 @@ test("a read-only participant's copied mode survives the durability flush and ma assert.equal((await stat(restored)).mode & 0o777, 0o444); }); +test("nested inventories retain depth-first sibling order and every byte and mode", async t => { + const {root, request} = await fixture(t); + const source = join(root, "tree"); + await mkdir(source, {mode: 0o700}); + await mkdir(join(source, "a"), {mode: 0o700}); + await mkdir(join(source, "a/b"), {mode: 0o700}); + const leaves = ["a/b/deep", "a/b-", "a/b.txt", "a-", "a.txt", + ...Array.from({length: 64}, (_, i) => `z${String(i).padStart(2, "0")}`)]; + for (const [i, name] of [...leaves].reverse().entries()) { + await writeFile(join(source, name), `evidence:${name}\n`, {mode: i % 2 ? 0o600 : 0o400}); + } + // The filesystem owns its supported permission bits (Windows exposes a + // different subset). Observe those independently before calling capture. + const modes = new Map(); + for (const name of ["", "a", "a/b", ...leaves]) modes.set(name, (await stat(join(source, name))).mode & 0o777); + const file = (name: string) => ({path: name, kind: "file", mode: modes.get(name)!, + size: Buffer.byteLength(`evidence:${name}\n`), sha256: createHash("sha256").update(`evidence:${name}\n`).digest("hex")}); + const directory = (name: string) => ({path: name, kind: "directory", mode: modes.get(name)!}); + // Flat string sorting would put a- before a/b; completion order is also + // irrelevant. This is the existing manifest's depth-first traversal contract. + const expected = [directory(""), directory("a"), directory("a/b"), ...leaves.map(file)]; + const components = [{id: "tree", source_path: source}]; + const probe = await checkpointArtifacts({action: "probe", components}); + assert.deepEqual(probe.source_witness, {tree: expected}); + const captured = await checkpointArtifacts({...request, components, expected_source_sha256: probe.source_sha256}); + const manifest = JSON.parse(await readFile(join(request.destination, "checkpoint.json"), "utf8")); + assert.deepEqual(manifest.source_witness, {tree: expected}); + assert.deepEqual(manifest.artifact_witness, {tree: expected}); + const restored = join(root, "restored"); + await checkpointArtifacts({action: "materialize", root: request.destination, + expected_sha256: captured.sha256, destination: restored}); + for (const name of leaves) { + const copy = join(restored, "components/tree", name); + assert.equal(await readFile(copy, "utf8"), `evidence:${name}\n`); + assert.equal((await stat(copy)).mode & 0o777, file(name).mode); + } +}); + test("changed source, duplicate identities and overlapping sources publish no unit", async t => { const {root, request} = await fixture(t); await writeFile(join(root, "conversation"), "new native cursor"); @@ -65,6 +106,48 @@ test("changed source, duplicate identities and overlapping sources publish no un assert.deepEqual((await readdir(root)).sort(), ["code", "conversation"]); }); +test("copied bytes reject ABA source movement even when the final source matches the probe", async t => { + const {root, request} = await fixture(t), original = fs.copyFile; + const source = join(root, "code/solution.sh"), bytes = await readFile(source); + const copy = t.mock.method(fs, "copyFile", async (...args: Parameters) => { + if (args[0] !== source) return original(...args); + await writeFile(source, "unreviewed intermediate bytes\n"); + try {await original(...args);} + finally {await writeFile(source, bytes);} + }); + syncBuiltinESMExports(); + try { + await assert.rejects(checkpointArtifacts(request), /copied bytes do not match/); + assert.deepEqual(await readFile(source), bytes); + assert.deepEqual((await readdir(root)).sort(), ["code", "conversation"]); + } finally {copy.mock.restore(); syncBuiltinESMExports();} +}); + +test("a failed copy drains started member IO before removing the unpublished staging tree", async t => { + const {root, request} = await fixture(t), original = fs.copyFile; + for (let i = 0; i < 24; i++) await writeFile(join(root, "code", `member-${i}`), `retained evidence ${i}`); + const probe = await checkpointArtifacts({action: "probe", components: request.components}); + let active = 0, peak = 0, calls = 0; + const copy = t.mock.method(fs, "copyFile", async (...args: Parameters) => { + active++; peak = Math.max(peak, active); + const first = calls++ === 0; + try { + // Controlled storage latency lets sibling copies overlap the injected + // failure. It is a cleanup invariant, not a performance threshold. + await new Promise(resolve => setTimeout(resolve, first ? 30 : 80)); + if (first) throw new Error("injected member copy failure"); + await original(...args); + } finally {active--;} + }); + syncBuiltinESMExports(); + try { + await assert.rejects(checkpointArtifacts({...request, expected_source_sha256: probe.source_sha256}), /injected member copy failure/); + assert.equal(active, 0, "no member IO may outlive staging cleanup"); + assert.ok(peak <= 8, "member IO must remain bounded"); + assert.deepEqual((await readdir(root)).sort(), ["code", "conversation"]); + } finally {copy.mock.restore(); syncBuiltinESMExports();} +}); + test("missing, changed and extra artifact members or foreign expected digest reject", async t => { const {root, request} = await fixture(t), captured = await checkpointArtifacts(request); const verify = {action: "verify", root: request.destination, expected_sha256: captured.sha256};