Stream controller validation digests within the server heap limit

Co-Authored-By: Paperclip <noreply@paperclip.ing>

server/src/services/native-runtime/native-controller-archive-source.ts

@@ -3,7 +3,7 @@ import { createHash } from "node:crypto";
 import { agentTaskSessions, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db";
 import { nativeSha256 } from "./canonical.js";
 import { findNativeSessionResumeRun } from "./native-session-resume.js";
-import { readLegacyNativeControllerArchiveSource } from "./native-controller-migration.js";
+import { readLegacyNativeControllerArchiveDigests, readLegacyNativeControllerArchiveSource } from "./native-controller-migration.js";
 import { tryLockNativeWorkspaceFinalization } from "./native-workspace-finalization-ownership.js";
 import { archiveFailedLegacyNativeController } from "./native-controller-archive.js";
 import type { StorageService } from "../../storage/types.js";
@@ -59,9 +59,9 @@ export async function assertSettledControllerArchiveSource(db: Db, owner: Settle
   const selected = await findNativeSessionResumeRun(db, { companyId: owner.companyId, agentId: owner.agentId, issueId: owner.issueId,
     normalizedSessionId: owner.sessionId, beforeCreatedAt: selectedRun.createdAt, includeProgressBarrier: true });
   if (!selected || selected.id !== owner.runId || nativeControllerArchiveProfileDigest(selected.runnerProfileJson) !== owner.profileDigest) fail();
-  const material = await readLegacyNativeControllerArchiveSource(selected);
-  if (material.controller === undefined || fileDigest(material.controller) !== owner.controllerDigest
-    || (material.supervisor === undefined ? null : fileDigest(material.supervisor)) !== owner.supervisorDigest) fail();
+  const material = await readLegacyNativeControllerArchiveDigests(selected);
+  if (material.controllerDigest !== owner.controllerDigest
+    || (material.supervisorDigest ?? null) !== owner.supervisorDigest) fail();
 }
 
 export async function archiveSettledFailedNativeController(db: Db, input: {

server/src/services/native-runtime/native-controller-migration.ts

@@ -18,6 +18,7 @@ import { durableNativeHarnessStorage } from "./native-harness-backup-storage.js"
 import { archiveFailedLegacyNativeController, hasArchivedLegacyNativeController } from "./native-controller-archive.js";
 import { redactRunnerApiPayloads } from "./runner-api-log-redaction.js";
 import { archiveSettledFailedNativeController } from "./native-controller-archive-source.js";
+import { nativeControllerSourceDigest } from "./native-controller-source-digest.js";
 
 type Selected = { id: string; companyId: string; agentId: string; issueId: string; sessionId: string; createdAt: Date; lastRunId: string };
 
@@ -135,6 +136,13 @@ export async function readLegacyNativeControllerArchiveSource(source: NonNullabl
   return { paths, controller, supervisor };
 }
 
+export async function readLegacyNativeControllerArchiveDigests(source: NonNullable<Awaited<ReturnType<typeof findNativeSessionResumeRun>>>) {
+  const paths = await localPaths(source);
+  return { controllerDigest: await nativeControllerSourceDigest(paths.controlFile, 256 * 1024 * 1024),
+    supervisorDigest: paths.execution.session.lifecyclePolicy.mode === "warm"
+      ? await nativeControllerSourceDigest(paths.supervisorPath, 4 * 1024 * 1024) : undefined };
+}
+
 function preservedFailedCheckpoint(source: NonNullable<Awaited<ReturnType<typeof findNativeSessionResumeRun>>>,
   sessionId: string, checkpoint: Awaited<ReturnType<typeof loadNativeRunnerCheckpoint>>) {
   return source.status === "failed" && source.runnerProfileJson.sessionCheckpoint == null

server/src/services/native-runtime/native-controller-source-digest.test.ts

@@ -0,0 +1,42 @@
+import { execFileSync } from "node:child_process";
+import { createHash } from "node:crypto";
+import { link, mkdtemp, open, rm, symlink, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { expect, it } from "vitest";
+import { nativeControllerSourceDigest } from "./native-controller-source-digest.js";
+
+it("hashes a maximum-size source repeatedly with an original copy retained on a 512 MiB heap", async () => {
+  const root = await mkdtemp(join(tmpdir(), "pc-controller-digest-"));
+  try {
+    const file = join(root, "controller.json"), chunk = Buffer.alloc(1024 * 1024, "x");
+    const expected = createHash("sha256"), handle = await open(file, "wx", 0o600);
+    try { for (let index = 0; index < 256; index++) { await handle.write(chunk); expected.update(chunk); } }
+    finally { await handle.close(); }
+    const source = new URL("./native-controller-source-digest.ts", import.meta.url).href;
+    const code = `import { readFile } from "node:fs/promises";
+      import { nativeControllerSourceDigest } from ${JSON.stringify(source)};
+      globalThis.retainedController = await readFile(${JSON.stringify(file)}, "utf8");
+      const hashes = [];
+      for (let index = 0; index < 3; index++) hashes.push(await nativeControllerSourceDigest(${JSON.stringify(file)}, 256 * 1024 * 1024));
+      console.log(JSON.stringify({ hashes, retainedBytes: globalThis.retainedController.length }));`;
+    const result = JSON.parse(execFileSync(process.execPath, ["--max-old-space-size=512", "--import", import.meta.resolve("tsx/esm"),
+      "--input-type=module", "--eval", code], { encoding: "utf8", timeout: 60_000, maxBuffer: 256 * 1024 }));
+    expect(result).toEqual({ hashes: Array(3).fill(expected.digest("hex")), retainedBytes: 256 * 1024 * 1024 });
+  } finally { await rm(root, { recursive: true, force: true }); }
+}, 60_000);
+
+it("keeps private-file and size guards while hashing", async () => {
+  const root = await mkdtemp(join(tmpdir(), "pc-controller-digest-guards-"));
+  try {
+    const file = join(root, "controller"); await writeFile(file, "source", { mode: 0o600 });
+    expect(await nativeControllerSourceDigest(join(root, "missing"), 10)).toBeUndefined();
+    expect(await nativeControllerSourceDigest(file, 10)).toBe(createHash("sha256").update("source").digest("hex"));
+    await expect(nativeControllerSourceDigest(file, 5)).rejects.toThrow("native_controller_migration_unsafe_source");
+    await expect(nativeControllerSourceDigest(root, 10)).rejects.toThrow("native_controller_migration_unsafe_source");
+    const alias = join(root, "alias"); await symlink(file, alias);
+    await expect(nativeControllerSourceDigest(alias, 10)).rejects.toMatchObject({ code: "ELOOP" });
+    await link(file, join(root, "hardlink"));
+    await expect(nativeControllerSourceDigest(file, 10)).rejects.toThrow("native_controller_migration_unsafe_source");
+  } finally { await rm(root, { recursive: true, force: true }); }
+});

server/src/services/native-runtime/native-controller-source-digest.ts

@@ -0,0 +1,23 @@
+import { constants } from "node:fs";
+import { open } from "node:fs/promises";
+import { createHash } from "node:crypto";
+
+export async function nativeControllerSourceDigest(file: string, maxBytes: number): Promise<string | undefined> {
+  const handle = await open(file, constants.O_RDONLY | constants.O_NOFOLLOW).catch((error: NodeJS.ErrnoException) => {
+    if (error.code === "ENOENT") return undefined;
+    throw error;
+  });
+  if (!handle) return undefined;
+  try {
+    const stat = await handle.stat();
+    if (!stat.isFile() || stat.nlink !== 1 || stat.size > maxBytes) throw new Error("native_controller_migration_unsafe_source");
+    const hash = createHash("sha256");
+    let bytes = 0;
+    for await (const chunk of handle.createReadStream({ autoClose: false, highWaterMark: 64 * 1024 })) {
+      bytes += chunk.length;
+      if (bytes > maxBytes) throw new Error("native_controller_migration_unsafe_source");
+      hash.update(chunk);
+    }
+    return hash.digest("hex");
+  } finally { await handle.close(); }
+}