feat(runner): persist controller authority and bounded delivery

Part 3 of the opt-in indexed history draft series. Validated at the integrated tip; activation remains disabled by default.

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

packages/plugins/sandbox-providers/daytona/src/file-sync-stream.test.ts

@@ -0,0 +1,27 @@
+import { mkdtemp, mkdir, readFile, rm, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { expect, it } from "vitest";
+import { visitCommandLines, writeTopLevelNames } from "./file-sync-stream.js";
+
+it("streams a listing beyond 32 MiB without retaining its lines", async () => {
+  let count = 0;
+  await visitCommandLines(process.execPath, ["-e", `const fs=require('fs'); for(let i=0;i<10000;i++) fs.writeSync(1,'x'.repeat(4096)+'\\n');`], line => { expect(line.length).toBe(4096); count++; });
+  expect(count).toBe(10_000);
+}, 30_000);
+
+it("bounds an individual entry and propagates child failure", async () => {
+  await expect(visitCommandLines(process.execPath, ["-e", "process.stdout.write('x'.repeat(70000))"], () => undefined)).rejects.toThrow("too long");
+  await expect(visitCommandLines(process.execPath, ["-e", "process.exit(7)"], () => undefined)).rejects.toThrow("listing failed");
+});
+
+it("writes literal NUL-delimited names without shell expansion or a directory argv", async () => {
+  const root = await mkdtemp(join(tmpdir(), "daytona-catalog-"));
+  try {
+    const source = join(root, "source"), names = join(root, "names"); await mkdir(source);
+    const entries = ["-C", "a name", "line\nbreak", "$(not-a-command)"];
+    for (const name of entries) await writeFile(join(source, name), "x");
+    await writeTopLevelNames(source, names);
+    expect((await readFile(names, "utf8")).split("\0").filter(Boolean).sort()).toEqual(entries.map(name => `./${name}`).sort());
+  } finally { await rm(root, { recursive: true, force: true }); }
+});

packages/plugins/sandbox-providers/daytona/src/file-sync-stream.ts

@@ -0,0 +1,37 @@
+import { spawn } from "node:child_process";
+import { open, opendir } from "node:fs/promises";
+
+/** NUL-delimited disk catalog; filenames never become a growing argv array. */
+export async function writeTopLevelNames(directory: string, destination: string): Promise<void> {
+  const file = await open(destination, "wx", 0o600);
+  try {
+    for await (const entry of await opendir(directory, { bufferSize: 128 })) await file.writeFile(`./${entry.name}\0`);
+  } finally { await file.close(); }
+}
+
+/** Stream a tar listing with a per-entry bound instead of a total listing cap. */
+export async function visitCommandLines(command: string, args: string[], visit: (line: string) => void): Promise<void> {
+  const child = spawn(command, args, { env: { ...process.env, COPYFILE_DISABLE: "1", LC_ALL: "C" }, stdio: ["ignore", "pipe", "pipe"] });
+  let pending = "", diagnosticBytes = 0;
+  const maxLine = 64 * 1024;
+  const finished = new Promise<void>((resolve, reject) => {
+    child.once("error", reject);
+    child.once("close", code => code === 0 ? resolve() : reject(new Error("Daytona sync archive listing failed")));
+  });
+  void finished.catch(() => undefined);
+  child.stderr.on("data", (chunk: Buffer) => { diagnosticBytes += chunk.length; if (diagnosticBytes > maxLine) child.kill(); });
+  child.stdout.setEncoding("utf8");
+  try {
+    for await (const chunk of child.stdout) {
+      pending += String(chunk); let end: number;
+      while ((end = pending.indexOf("\n")) >= 0) {
+        if (end > maxLine) throw new Error("Daytona sync archive entry listing too long");
+        visit(pending.slice(0, end)); pending = pending.slice(end + 1);
+      }
+      if (pending.length > maxLine) throw new Error("Daytona sync archive entry listing too long");
+    }
+    if (pending) visit(pending);
+    await finished;
+    if (diagnosticBytes > maxLine) throw new Error("Daytona sync archive listing failed");
+  } catch (error) { child.kill(); await finished.catch(() => undefined); throw error; }
+}

packages/plugins/sandbox-providers/daytona/src/file-sync.ts

@@ -1,3 +1,4 @@
+import { visitCommandLines, writeTopLevelNames } from "./file-sync-stream.js";
 import path from "node:path";
 import os from "node:os";
 import { promises as fs, createReadStream, createWriteStream } from "node:fs";
@@ -164,32 +165,13 @@ async function createHostTarball(input: {
   followSymlinks?: boolean;
 }): Promise<void> {
   const excludeArgs = ["._*", ...(input.exclude ?? [])].flatMap((entry) => ["--exclude", entry]);
-  const entries = (await fs.readdir(input.localDir)).sort((left, right) => left.localeCompare(right));
-  if (entries.length === 0) {
-    // An empty source is valid (blank workspace / empty asset dir). Write a valid
-    // gzip-compressed empty tar (1024-byte zero EOF marker) so extraction is a
-    // clean no-op and uses the same transport as non-empty directories.
-    await fs.writeFile(input.archivePath, await new Promise<Buffer>((resolve, reject) => {
-      zlib.gzip(Buffer.alloc(1024), (error, compressed) => error ? reject(error) : resolve(compressed));
-    }));
-    return;
-  }
-  await execFileAsync(
-    "tar",
-    [
-      "-cz",
-      "--no-xattrs",
-      ...(input.followSymlinks ? ["-h"] : []),
-      "-f",
-      input.archivePath,
-      "-C",
-      input.localDir,
-      ...excludeArgs,
-      "--",
-      ...entries,
-    ],
-    { env: { ...process.env, COPYFILE_DISABLE: "1" }, maxBuffer: 32 * 1024 * 1024 },
-  );
+  const namesPath = `${input.archivePath}.names`;
+  try {
+    await writeTopLevelNames(input.localDir, namesPath);
+    await execFileAsync("tar", ["-cz", "--no-xattrs", ...(input.followSymlinks ? ["-h"] : []),
+      "-f", input.archivePath, "-C", input.localDir, ...excludeArgs, "--null", "-T", namesPath],
+      { env: { ...process.env, COPYFILE_DISABLE: "1" }, maxBuffer: 64 * 1024 });
+  } finally { await fs.rm(namesPath, { force: true }); }
 }
 
 /**
@@ -251,12 +233,8 @@ export function splitLinkEntryOnce(field: string, delimiter: string): { name: st
  * targets are inspected; any unparseable line fails closed.
  */
 async function assertTarballEntriesConfined(archivePath: string): Promise<void> {
-  const { stdout } = await execFileAsync("tar", ["-tvf", archivePath], {
-    env: { ...process.env, COPYFILE_DISABLE: "1" },
-    maxBuffer: 32 * 1024 * 1024,
-  });
-  const lines = stdout.split("\n").filter((line) => line.trim().length > 0);
-  for (const line of lines) {
+  await visitCommandLines("tar", ["-tvf", archivePath], (line) => {
+    if (!line.trim()) return;
     const parsed = parseTarVerboseListingLine(line);
     if (!parsed) {
       throw new Error(`Daytona syncOut refusing tarball with an unparseable entry listing: ${line}`);
@@ -287,7 +265,7 @@ async function assertTarballEntriesConfined(archivePath: string): Promise<void>
         );
       }
     }
-  }
+  });
 }
 
 async function extractHostTarball(input: { archivePath: string; localDir: string }): Promise<void> {
@@ -305,8 +283,7 @@ async function countHostFiles(root: string, exclude?: string[]): Promise<number>
   const excludeSet = new Set(exclude ?? []);
   let total = 0;
   const walk = async (dir: string): Promise<void> => {
-    const entries = await fs.readdir(dir, { withFileTypes: true }).catch(() => []);
-    for (const entry of entries) {
+    for await (const entry of await fs.opendir(dir, { bufferSize: 128 })) {
       if (excludeSet.has(entry.name)) continue;
       const full = path.join(dir, entry.name);
       if (entry.isDirectory()) {
@@ -1183,14 +1160,14 @@ async function syncOutDirectoryMapping(input: {
     // Tar the source in-sandbox (naming top-level entries so no "." self-entry is
     // embedded), reproducing the `followSymlinks` → `-h` mapping, then stream the
     // single archive back over the native bulk channel.
+    const namesPath = `${remoteTar}.names`;
     const tarScript = [
+      "set -e",
       `cd ${shellQuote(mapping.sourcePath)}`,
-      "set -- *",
-      'if [ "$#" -eq 1 ] && [ "$1" = "*" ] && [ ! -e "$1" ] && [ ! -L "$1" ]; then set --; fi',
-      'for entry in .[!.]* ..?*; do [ -e "$entry" ] || [ -L "$entry" ] || continue; set -- "$@" "$entry"; done',
-      `if [ "$#" -eq 0 ]; then dd if=/dev/zero of=${shellQuote(remoteTar)} bs=1024 count=1; ` +
-        `else tar -c --no-xattrs ${mapping.followSymlinks ? "-h " : ""}${excludeFlags} -f ${shellQuote(remoteTar)} -- "$@"; fi`,
-    ].join(" && ");
+      `trap ${shellQuote(`rm -f -- ${shellQuote(namesPath)}`)} EXIT`,
+      `find . -mindepth 1 -maxdepth 1 -print0 > ${shellQuote(namesPath)}`,
+      `tar -c --no-xattrs ${mapping.followSymlinks ? "-h " : ""}${excludeFlags} -f ${shellQuote(remoteTar)} --null -T ${shellQuote(namesPath)}`,
+    ].join("; ");
     await assertSandboxCommandOk(sandbox, `sh -c ${shellQuote(tarScript)}`, timeoutSeconds, "syncOut tar");
     guardRoundTrips += 1;