fix: recover transient sandbox file transfer failures safely
Retry bounded read-only transport failures and resolve uncertain bulk writes through atomic signed receipts without replaying claimed batches. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Dotta
parent
ffb01ff3d5
commit
daf1d64b20
|
|
@ -223,6 +223,19 @@ Other providers keep the small-argument transport. Incoming storage responses
|
|||
are prefetched four at a time and closed if transfer fails. Repository restores
|
||||
use the same path, then recreate confined repository links.
|
||||
|
||||
Read-only sandbox commands retry transient connection failures and HTTP
|
||||
502/503/504 responses up to three attempts within one 120-second deadline.
|
||||
Script failures, invalid responses, and authorization failures are not retried.
|
||||
An incoming bulk batch has a unique ID and a signed receipt in its private
|
||||
staging directory. If the provider loses the response, the host checks that
|
||||
receipt: a completed batch is acknowledged without publishing its files again;
|
||||
only a missing claim permits resubmitting the same batch ID. An atomic claim
|
||||
prevents two concurrent submissions from applying the same batch twice. A
|
||||
running or interrupted batch is never replayed. Its bounded outcome check
|
||||
either observes completion or fails visibly and retains the working copy for
|
||||
the next run's existing intent reconciliation. This does not make arbitrary
|
||||
sandbox commands or repository mutations retryable.
|
||||
|
||||
Repository checkpoints transfer at most four distinct content-addressed blobs
|
||||
concurrently, avoiding duplicate uploads for identical files. Small-file reads
|
||||
are grouped into at most 1 MiB and 64 files per remote command, with at most
|
||||
|
|
|
|||
|
|
@ -0,0 +1,167 @@
|
|||
import { afterEach, describe, expect, it } from "vitest";
|
||||
import { mkdtemp, mkdir, readFile, realpath, rm, writeFile, symlink, link, readdir } from "node:fs/promises";
|
||||
import { spawn, type ChildProcess } from "node:child_process";
|
||||
import { createHash, randomBytes, randomUUID } from "node:crypto";
|
||||
import { setTimeout as delay } from "node:timers/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { localTestWorkFolderRunner } from "./helpers/work-folder-runner.js";
|
||||
|
||||
const source = await readFile(new URL("../services/scripts/work-folder-io.mjs", import.meta.url), "utf8");
|
||||
const sha = (body: string) => createHash("sha256").update(body).digest("hex");
|
||||
type Envelope = { operation: string; root: string; stagingRoot: string; batchId: string; batchSha256: string; batchReceiptKey: string };
|
||||
const roots: string[] = [], children = new Set<ChildProcess>();
|
||||
async function fixture(operations: Record<string, unknown>[] = [{ operation: "write", path: "piece", offset: 0, data: Buffer.from("old").toString("base64") },
|
||||
{ operation: "publish", path: "file", stagingPath: "piece", sha256: sha("old"), executable: false }]) {
|
||||
const directory = await realpath(await mkdtemp(path.join(os.tmpdir(), "work-folder-receipt-")));
|
||||
roots.push(directory);
|
||||
const root = path.join(directory, "task"), stagingRoot = path.join(directory, "staging");
|
||||
await mkdir(root); await mkdir(stagingRoot);
|
||||
const body = JSON.stringify(operations);
|
||||
const request: Envelope = { operation: "batch", root, stagingRoot, batchId: randomUUID(), batchSha256: sha(body), batchReceiptKey: randomBytes(32).toString("hex") };
|
||||
return { directory, request, body, receipt: path.join(stagingRoot, ".batch-receipts", request.batchId) };
|
||||
}
|
||||
async function invoke(request: Envelope, body?: string) {
|
||||
return localTestWorkFolderRunner.execute({ command: process.execPath,
|
||||
args: ["--input-type=module", "-e", source, Buffer.from(JSON.stringify(request)).toString("base64")],
|
||||
...(body === undefined ? {} : { stdin: body }), timeoutMs: 5000 });
|
||||
}
|
||||
async function status(request: Envelope) {
|
||||
const result = await invoke({ ...request, operation: "batch-status" });
|
||||
expect(result.exitCode, result.stderr).toBe(0);
|
||||
return JSON.parse(result.stdout);
|
||||
}
|
||||
async function successful(request: Envelope, body: string) {
|
||||
const result = await invoke(request, body);
|
||||
expect(result.exitCode, result.stderr).toBe(0);
|
||||
return JSON.parse(result.stdout);
|
||||
}
|
||||
afterEach(async () => {
|
||||
for (const child of children) {
|
||||
if (child.exitCode !== null || child.signalCode !== null) continue;
|
||||
const closed = new Promise<void>(resolve => child.once("close", () => resolve()));
|
||||
child.kill("SIGKILL"); await closed;
|
||||
}
|
||||
children.clear();
|
||||
for (const directory of roots.splice(0)) await rm(directory, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
describe("bulk work-folder receipts in the actual sandbox IO script", () => {
|
||||
it("reports missing without creating receipt directories", async () => {
|
||||
const f = await fixture();
|
||||
expect(await status(f.request)).toEqual({ state: "missing" });
|
||||
expect(await readdir(f.request.stagingRoot)).toEqual([]);
|
||||
});
|
||||
it("returns the original count after a lost response without overwriting a later edit", async () => {
|
||||
const f = await fixture();
|
||||
// Discarding this response models a provider response lost after execution.
|
||||
await successful(f.request, f.body);
|
||||
await writeFile(path.join(f.request.root, "file"), "later agent edit");
|
||||
expect(await status(f.request)).toEqual({ state: "completed", completed: 2 });
|
||||
expect(await successful(f.request, f.body)).toEqual({ completed: 2 });
|
||||
expect(await readFile(path.join(f.request.root, "file"), "utf8")).toBe("later agent edit");
|
||||
expect(await readdir(f.request.stagingRoot)).toEqual([".batch-receipts"]);
|
||||
expect(await readFile(f.receipt, "utf8")).not.toContain(f.request.batchReceiptKey);
|
||||
});
|
||||
it("atomically claims concurrent duplicate submissions", async () => {
|
||||
const f = await fixture();
|
||||
const outcomes = await Promise.all([successful(f.request, f.body), successful(f.request, f.body)]);
|
||||
expect(outcomes.some(value => value.completed === 2)).toBe(true);
|
||||
expect(outcomes.every(value => value.completed === 2 || value.pending === true)).toBe(true);
|
||||
expect(await status(f.request)).toEqual({ state: "completed", completed: 2 });
|
||||
expect(await readFile(path.join(f.request.root, "file"), "utf8")).toBe("old");
|
||||
});
|
||||
it("never replays a running or crashed batch after its exclusive claim", async () => {
|
||||
const f = await fixture(), marker = path.join(f.directory, "entered-operation");
|
||||
// Instrument only the subprocess test source: stop after the real signed
|
||||
// claim, before the first real write. Production has no fault/test hook.
|
||||
const stalled = source.replace("function execute(request) {", `function execute(request) {\nif (request.operation === "write") { fs.writeFileSync(${JSON.stringify(marker)}, "entered"); Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0); }`);
|
||||
const child = spawn(process.execPath, ["--input-type=module", "-e", stalled, Buffer.from(JSON.stringify(f.request)).toString("base64")], { stdio: ["pipe", "ignore", "ignore"] });
|
||||
children.add(child); child.stdin.end(f.body);
|
||||
const end = Date.now() + 5000;
|
||||
while (true) { try { await readFile(marker); break; } catch { if (Date.now() > end) throw new Error("Child did not reach mutation barrier"); await delay(10); } }
|
||||
expect(await status(f.request)).toEqual({ state: "running" });
|
||||
expect(await successful(f.request, f.body)).toEqual({ pending: true });
|
||||
const closed = new Promise<void>(resolve => child.once("close", () => resolve())); child.kill("SIGKILL"); await closed; children.delete(child);
|
||||
expect(await status(f.request)).toEqual({ state: "running" });
|
||||
expect(await successful(f.request, f.body)).toEqual({ pending: true });
|
||||
expect(await readdir(f.request.root)).toEqual([]);
|
||||
});
|
||||
it("treats an empty interrupted claim as pending, never completed or replayable", async () => {
|
||||
const f = await fixture(); await mkdir(path.dirname(f.receipt)); await writeFile(f.receipt, "");
|
||||
expect(await status(f.request)).toEqual({ state: "running" });
|
||||
expect(await successful(f.request, f.body)).toEqual({ pending: true });
|
||||
expect(await readdir(f.request.root)).toEqual([]);
|
||||
});
|
||||
it("verifies exact stdin bytes before creating a claim or mutating files", async () => {
|
||||
const f = await fixture(); const result = await invoke(f.request, f.body + " ");
|
||||
expect(result.exitCode).not.toBe(0); expect(result.stderr).toContain("batch_body_hash_mismatch");
|
||||
expect(await status(f.request)).toEqual({ state: "missing" });
|
||||
expect(await readdir(f.request.root)).toEqual([]);
|
||||
});
|
||||
it("rejects the same ID with different body, root, or signing key", async () => {
|
||||
const f = await fixture(); await successful(f.request, f.body);
|
||||
const body = JSON.stringify([{ operation: "mkdir", path: "other" }]);
|
||||
const different = await invoke({ ...f.request, batchSha256: sha(body) }, body);
|
||||
expect(different.exitCode).not.toBe(0); expect(different.stderr).toContain("batch_receipt_identity_mismatch");
|
||||
const otherRoot = path.join(f.directory, "other"); await mkdir(otherRoot);
|
||||
for (const request of [{ ...f.request, root: otherRoot }, { ...f.request, batchReceiptKey: randomBytes(32).toString("hex") }]) {
|
||||
expect((await invoke({ ...request, operation: "batch-status" })).exitCode).not.toBe(0);
|
||||
}
|
||||
expect(await readFile(path.join(f.request.root, "file"), "utf8")).toBe("old");
|
||||
expect(await readdir(otherRoot)).toEqual([]);
|
||||
});
|
||||
it("binds receipts to staging root as well as task root", async () => {
|
||||
const f = await fixture(); await successful(f.request, f.body);
|
||||
const stagingRoot = path.join(f.directory, "other-stage"); await mkdir(path.join(stagingRoot, ".batch-receipts"), { recursive: true });
|
||||
await writeFile(path.join(stagingRoot, ".batch-receipts", f.request.batchId), await readFile(f.receipt));
|
||||
const result = await invoke({ ...f.request, operation: "batch-status", stagingRoot });
|
||||
expect(result.exitCode).not.toBe(0); expect(result.stderr).toContain("batch_receipt_identity_mismatch");
|
||||
});
|
||||
it.each(["completed", "extra", "signature", "malformed", "oversized"])("rejects forged or corrupted %s receipts", async corruption => {
|
||||
const f = await fixture(); await successful(f.request, f.body);
|
||||
const receipt = JSON.parse(await readFile(f.receipt, "utf8"));
|
||||
if (corruption === "completed") receipt.completed = 1;
|
||||
if (corruption === "extra") receipt.ignored = true;
|
||||
if (corruption === "signature") receipt.signature = "0".repeat(64);
|
||||
await writeFile(f.receipt, corruption === "malformed" ? "{" : corruption === "oversized" ? " ".repeat(5000) : JSON.stringify(receipt));
|
||||
const result = await invoke({ ...f.request, operation: "batch-status" });
|
||||
expect(result.exitCode).not.toBe(0); expect(result.stderr).toContain("invalid_batch_receipt");
|
||||
expect((await invoke(f.request, f.body)).exitCode).not.toBe(0);
|
||||
});
|
||||
it("records failure after a partial batch and refuses replay even if the cause is repaired", async () => {
|
||||
const operations = [{ operation: "write", path: "piece", offset: 0, data: Buffer.from("old").toString("base64") },
|
||||
{ operation: "publish", path: "file", stagingPath: "piece", sha256: sha("wrong"), executable: false }];
|
||||
const f = await fixture(operations); const failure = await invoke(f.request, f.body);
|
||||
expect(failure.exitCode).not.toBe(0); expect(failure.stderr).toContain("content_changed_during_transfer");
|
||||
expect(await status(f.request)).toEqual({ state: "failed", error: "batch_execution_failed" });
|
||||
await writeFile(path.join(f.request.stagingRoot, "piece"), "wrong");
|
||||
expect((await invoke(f.request, f.body)).exitCode).not.toBe(0);
|
||||
expect(await readdir(f.request.root)).toEqual([]);
|
||||
expect(await readFile(path.join(f.request.stagingRoot, "piece"), "utf8")).toBe("wrong");
|
||||
});
|
||||
it.each(["receipt-symlink", "receipt-hardlink", "parent-symlink", "staging-symlink"])("rejects %s without touching the target", async variant => {
|
||||
const f = await fixture(), outside = path.join(f.directory, "outside"); await mkdir(outside);
|
||||
const secret = path.join(outside, "private"); await writeFile(secret, "untouched");
|
||||
let request = f.request;
|
||||
if (variant === "receipt-symlink" || variant === "receipt-hardlink") {
|
||||
await mkdir(path.dirname(f.receipt));
|
||||
if (variant === "receipt-symlink") await symlink(secret, f.receipt); else await link(secret, f.receipt);
|
||||
} else if (variant === "parent-symlink") await symlink(outside, path.dirname(f.receipt));
|
||||
else { const alias = path.join(f.directory, "stage-alias"); await symlink(f.request.stagingRoot, alias); request = { ...request, stagingRoot: alias }; }
|
||||
expect((await invoke({ ...request, operation: "batch-status" })).exitCode).not.toBe(0);
|
||||
expect((await invoke(request, f.body)).exitCode).not.toBe(0);
|
||||
expect(await readFile(secret, "utf8")).toBe("untouched");
|
||||
});
|
||||
it.each([
|
||||
{ operation: "write", path: ".batch-receipts/forged", offset: 0, data: "" },
|
||||
{ operation: "publish", path: "file", stagingPath: ".batch-receipts/forged", sha256: sha(""), executable: false },
|
||||
{ operation: "mkdir", path: "file", root: "/tmp" },
|
||||
{ operation: "write", path: "file", offset: 0, data: "", stagingRoot: "/tmp" },
|
||||
{ operation: "write", path: "../outside", offset: 0, data: "" },
|
||||
])("rejects reserved paths or body-authored roots before claiming: $operation $path", async operation => {
|
||||
const f = await fixture([operation]); expect((await invoke(f.request, f.body)).exitCode).not.toBe(0);
|
||||
expect(await status(f.request)).toEqual({ state: "missing" });
|
||||
expect(await readdir(f.request.root)).toEqual([]);
|
||||
});
|
||||
});
|
||||
|
|
@ -0,0 +1,96 @@
|
|||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { mkdtemp, mkdir, readFile, realpath, rm, writeFile } from "node:fs/promises";
|
||||
import { createHash } from "node:crypto";
|
||||
import { Readable } from "node:stream";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { workFolderTransport } from "../services/work-folder-transport.js";
|
||||
import { localTestWorkFolderRunner } from "./helpers/work-folder-runner.js";
|
||||
|
||||
const upstreamError = () => Object.assign(new Error("Request failed with status code 502"), { response: { status: 502 } });
|
||||
type ExecuteInput = Parameters<typeof localTestWorkFolderRunner.execute>[0];
|
||||
const request = (input: ExecuteInput) => JSON.parse(Buffer.from(input.args!.at(-1)!, "base64").toString());
|
||||
const entry = { path: "nested/file", kind: "file" as const, byteSize: 3,
|
||||
sha256: createHash("sha256").update("new").digest("hex"), executable: true };
|
||||
|
||||
describe("work folder batch outcome recovery", () => {
|
||||
const roots: string[] = [];
|
||||
async function fixture() {
|
||||
const root = await realpath(await mkdtemp(path.join(os.tmpdir(), "work-folder-receipt-host-")));
|
||||
roots.push(root);
|
||||
const target = path.join(root, "task"), staging = path.join(root, "staging");
|
||||
await mkdir(target); await mkdir(staging);
|
||||
return { target, staging };
|
||||
}
|
||||
afterEach(async () => {
|
||||
vi.useRealTimers(); vi.restoreAllMocks();
|
||||
for (const root of roots.splice(0)) await rm(root, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
it("recovers a lost successful batch response without overwriting a subsequent local edit", async () => {
|
||||
const { target, staging } = await fixture();
|
||||
const execute = vi.fn(async (input: ExecuteInput) => {
|
||||
const result = await localTestWorkFolderRunner.execute(input);
|
||||
if (request(input).operation === "batch") {
|
||||
expect(result.exitCode).toBe(0);
|
||||
expect(await readFile(path.join(target, entry.path), "utf8")).toBe("new");
|
||||
await writeFile(path.join(target, entry.path), "later");
|
||||
throw upstreamError();
|
||||
}
|
||||
return result;
|
||||
});
|
||||
const transport = workFolderTransport({ execute, supportsSingleStreamStdinProgress: true });
|
||||
const beforePublish = vi.fn();
|
||||
await transport.writeMany(target, staging, (async function* () { yield { entry, body: Readable.from(["new"]) }; })(), beforePublish);
|
||||
expect(execute.mock.calls.map(([input]) => request(input).operation)).toEqual(["batch", "batch-status"]);
|
||||
expect(beforePublish).toHaveBeenCalledTimes(1);
|
||||
expect(await readFile(path.join(target, entry.path), "utf8")).toBe("later");
|
||||
});
|
||||
|
||||
it("resubmits the same identity only when a lost request left no claim", async () => {
|
||||
const { target, staging } = await fixture();
|
||||
let initial = true;
|
||||
const execute = vi.fn(async (input: ExecuteInput) => {
|
||||
if (request(input).operation === "batch" && initial) { initial = false; throw upstreamError(); }
|
||||
return localTestWorkFolderRunner.execute(input);
|
||||
});
|
||||
await workFolderTransport({ execute, supportsSingleStreamStdinProgress: true })
|
||||
.write(target, staging, entry, Readable.from(["new"]));
|
||||
const requests = execute.mock.calls.map(([input]) => request(input));
|
||||
expect(requests.map((r) => r.operation)).toEqual(["batch", "batch-status", "batch"]);
|
||||
expect(new Set(requests.map((r) => r.batchId)).size).toBe(1);
|
||||
expect(new Set(requests.map((r) => r.batchReceiptKey)).size).toBe(1);
|
||||
expect(new Set(requests.map((r) => r.batchSha256)).size).toBe(1);
|
||||
expect(await readFile(path.join(target, entry.path), "utf8")).toBe("new");
|
||||
});
|
||||
|
||||
it("preserves a claimed incomplete batch instead of retrying its mutations", async () => {
|
||||
vi.useFakeTimers();
|
||||
const execute = vi.fn(async (input: ExecuteInput) => ({ exitCode: 0, stderr: "", signal: null, timedOut: false,
|
||||
stdout: JSON.stringify(request(input).operation === "batch" ? { pending: true } : { state: "running" }) }));
|
||||
const pending = workFolderTransport({ execute, supportsSingleStreamStdinProgress: true })
|
||||
.write("/task", "/staging", entry, Readable.from(["new"]));
|
||||
const rejection = expect(pending).rejects.toThrow("outcome is uncertain");
|
||||
await vi.advanceTimersByTimeAsync(120_000);
|
||||
await rejection;
|
||||
expect(execute.mock.calls.filter(([input]) => request(input).operation === "batch")).toHaveLength(1);
|
||||
expect(execute.mock.calls.length).toBeLessThanOrEqual(242);
|
||||
});
|
||||
|
||||
it("bounds absent-claim retries and never retries a recorded failure", async () => {
|
||||
const execute = vi.fn(async (input: ExecuteInput) => {
|
||||
if (request(input).operation === "batch") throw upstreamError();
|
||||
return { exitCode: 0, stderr: "", signal: null, timedOut: false, stdout: JSON.stringify({ state: "missing" }) };
|
||||
});
|
||||
const transport = workFolderTransport({ execute, supportsSingleStreamStdinProgress: true });
|
||||
await expect(transport.write("/task", "/staging", entry, Readable.from(["new"]))).rejects.toThrow("502");
|
||||
expect(execute.mock.calls.filter(([input]) => request(input).operation === "batch")).toHaveLength(3);
|
||||
execute.mockClear();
|
||||
execute.mockImplementation(async (input: ExecuteInput) => {
|
||||
if (request(input).operation === "batch") throw upstreamError();
|
||||
return { exitCode: 0, stderr: "", signal: null, timedOut: false, stdout: JSON.stringify({ state: "failed", error: "batch_failed" }) };
|
||||
});
|
||||
await expect(transport.write("/task", "/staging", entry, Readable.from(["new"]))).rejects.toThrow("batch_failed");
|
||||
expect(execute.mock.calls.map(([input]) => request(input).operation)).toEqual(["batch", "batch-status"]);
|
||||
});
|
||||
});
|
||||
|
|
@ -0,0 +1,142 @@
|
|||
import { createHash } from "node:crypto";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime";
|
||||
import { JsonRpcCallError } from "@paperclipai/plugin-sdk/protocol";
|
||||
import { workFolderTransport } from "../services/work-folder-transport.js";
|
||||
|
||||
const clock = vi.hoisted(() => ({ now: 0, delay: vi.fn() }));
|
||||
vi.mock("node:timers/promises", async (original) => ({
|
||||
...await original<typeof import("node:timers/promises")>(),
|
||||
setTimeout: clock.delay,
|
||||
}));
|
||||
|
||||
type Command = Parameters<CommandManagedRuntimeRunner["execute"]>[0];
|
||||
type Transport = ReturnType<typeof workFolderTransport>;
|
||||
const root = "/home/daytona/task";
|
||||
const bytes = Buffer.from("verified file bytes\n");
|
||||
const entry = { path: "nested/file", kind: "file" as const, byteSize: bytes.length,
|
||||
sha256: createHash("sha256").update(bytes).digest("hex"), executable: true };
|
||||
const request = (input: Command) => JSON.parse(Buffer.from(input.args!.at(-1)!, "base64").toString());
|
||||
const result = (value: unknown) => ({ stdout: JSON.stringify(value), stderr: "", exitCode: 0, signal: null, timedOut: false });
|
||||
const operations = ["scan", "read", "read-batch"] as const;
|
||||
type Operation = typeof operations[number];
|
||||
|
||||
function successfulCommand(input: Command) {
|
||||
const decoded = request(input);
|
||||
expect(decoded.root).toBe(root);
|
||||
if (decoded.operation === "scan") return result([entry]);
|
||||
if (decoded.operation === "read") {
|
||||
expect(decoded.path).toBe(entry.path);
|
||||
return result({ data: bytes.subarray(decoded.offset, decoded.offset + decoded.length).toString("base64") });
|
||||
}
|
||||
expect(decoded.operation).toBe("read-batch");
|
||||
expect(decoded.entries).toEqual([{ path: entry.path, byteSize: bytes.length }]);
|
||||
return result([{ data: bytes.toString("base64") }]);
|
||||
}
|
||||
|
||||
async function consume(transport: Transport, operation: Operation) {
|
||||
if (operation === "scan") return transport.scan(root);
|
||||
if (operation === "read-batch") return transport.readBatch!(root, [entry]);
|
||||
const chunks: Buffer[] = [];
|
||||
for await (const chunk of transport.read(root, entry.path, bytes.length)) chunks.push(Buffer.from(chunk));
|
||||
return Buffer.concat(chunks);
|
||||
}
|
||||
|
||||
const transientErrors = [
|
||||
{ label: "typed HTTP 502", make: () => Object.assign(new Error("provider unavailable"), { response: { status: 502 } }) },
|
||||
{ label: "typed HTTP 503", make: () => Object.assign(new Error("provider unavailable"), { status: 503 }) },
|
||||
{ label: "typed HTTP 504", make: () => Object.assign(new Error("provider unavailable"), { statusCode: 504 }) },
|
||||
{ label: "staging plugin RPC HTTP 502", make: () => new JsonRpcCallError({ code: -32002,
|
||||
message: "Request failed with status code 502: Sandbox command requested here" }) },
|
||||
];
|
||||
|
||||
describe("work folder read transport retry boundaries", () => {
|
||||
beforeEach(() => {
|
||||
clock.now = 1_000_000;
|
||||
clock.delay.mockReset().mockImplementation(async (ms: number) => { clock.now += ms; });
|
||||
vi.spyOn(Date, "now").mockImplementation(() => clock.now);
|
||||
});
|
||||
afterEach(() => { vi.restoreAllMocks(); });
|
||||
|
||||
describe.each(operations)("%s", (operation) => {
|
||||
it.each(transientErrors)("recovers $label without changing the read or duplicating bytes", async ({ make }) => {
|
||||
const execute = vi.fn<CommandManagedRuntimeRunner["execute"]>()
|
||||
.mockRejectedValueOnce(make()).mockRejectedValueOnce(make()).mockImplementation(async (input) => successfulCommand(input));
|
||||
const transport = workFolderTransport({ execute, supportsSingleStreamStdinProgress: true });
|
||||
const received = await consume(transport, operation);
|
||||
expect(received).toEqual(operation === "scan" ? [entry] : operation === "read-batch" ? [bytes] : bytes);
|
||||
expect(execute).toHaveBeenCalledTimes(3);
|
||||
expect(execute.mock.calls.map(([input]) => request(input))).toEqual(Array(3).fill(request(execute.mock.calls[0]![0])));
|
||||
expect(execute.mock.calls.map(([input]) => input.timeoutMs)).toEqual([120_000, 119_750, 119_250]);
|
||||
expect(clock.delay.mock.calls).toEqual([[250], [500]]);
|
||||
});
|
||||
|
||||
it("returns the last transport error after three failed attempts", async () => {
|
||||
const errors = transientErrors.slice(0, 3).map(({ make }) => make());
|
||||
const execute = vi.fn<CommandManagedRuntimeRunner["execute"]>()
|
||||
.mockRejectedValueOnce(errors[0]).mockRejectedValueOnce(errors[1]).mockRejectedValueOnce(errors[2]);
|
||||
await expect(consume(workFolderTransport({ execute, supportsSingleStreamStdinProgress: true }), operation)).rejects.toBe(errors[2]);
|
||||
expect(execute).toHaveBeenCalledTimes(3);
|
||||
expect(clock.delay).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("does not schedule a retry whose backoff exceeds the shared 120-second budget", async () => {
|
||||
const error = transientErrors[0]!.make();
|
||||
const execute = vi.fn<CommandManagedRuntimeRunner["execute"]>().mockImplementation(async () => {
|
||||
clock.now += 119_900;
|
||||
throw error;
|
||||
});
|
||||
await expect(consume(workFolderTransport({ execute, supportsSingleStreamStdinProgress: true }), operation)).rejects.toBe(error);
|
||||
expect(execute).toHaveBeenCalledTimes(1);
|
||||
expect(clock.delay).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
it("subtracts elapsed provider time from later attempts instead of resetting the deadline", async () => {
|
||||
const error = transientErrors[3]!.make();
|
||||
const execute = vi.fn<CommandManagedRuntimeRunner["execute"]>().mockImplementation(async (input) => {
|
||||
clock.now += execute.mock.calls.length === 1 ? 119_000 : input.timeoutMs!;
|
||||
throw error;
|
||||
});
|
||||
await expect(workFolderTransport({ execute }).scan(root)).rejects.toBe(error);
|
||||
expect(execute.mock.calls.map(([input]) => input.timeoutMs)).toEqual([120_000, 750]);
|
||||
expect(clock.now).toBe(1_120_000);
|
||||
expect(clock.delay.mock.calls).toEqual([[250]]);
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ label: "HTTP 401", error: Object.assign(new Error("denied"), { response: { status: 401 } }) },
|
||||
{ label: "HTTP 403", error: Object.assign(new Error("denied"), { status: 403 }) },
|
||||
{ label: "plain error with the staging text", error: new Error("Request failed with status code 502: Sandbox command requested here") },
|
||||
{ label: "different RPC error code", error: new JsonRpcCallError({ code: -32603, message: "Request failed with status code 502" }) },
|
||||
{ label: "arbitrary RPC command error containing 502", error: new JsonRpcCallError({ code: -32002, message: "script failed: Request failed with status code 502" }) },
|
||||
{ label: "unrecognized RPC suffix", error: new JsonRpcCallError({ code: -32002, message: "Request failed with status code 502: application rejected path" }) },
|
||||
{ label: "RPC authorization denial", error: new JsonRpcCallError({ code: -32002, message: "Request failed with status code 403: Sandbox command requested here" }) },
|
||||
])("does not retry $label", async ({ error }) => {
|
||||
const execute = vi.fn<CommandManagedRuntimeRunner["execute"]>().mockRejectedValue(error);
|
||||
await expect(workFolderTransport({ execute }).scan(root)).rejects.toBe(error);
|
||||
expect(execute).toHaveBeenCalledTimes(1);
|
||||
expect(clock.delay).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ label: "script failure", response: { ...result([]), exitCode: 1, stderr: "path_conflict: HTTP 502" } },
|
||||
{ label: "command timeout", response: { ...result([]), timedOut: true } },
|
||||
{ label: "invalid JSON", response: { ...result([]), stdout: "not json" } },
|
||||
{ label: "invalid scan paths", response: result([{ ...entry, path: "../outside" }]) },
|
||||
])("preserves $label as a non-retryable failure", async ({ response }) => {
|
||||
const execute = vi.fn<CommandManagedRuntimeRunner["execute"]>().mockResolvedValue(response);
|
||||
await expect(workFolderTransport({ execute }).scan(root)).rejects.toThrow();
|
||||
expect(execute).toHaveBeenCalledTimes(1);
|
||||
expect(clock.delay).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("does not replay a root move after a typed HTTP 503 with an uncertain outcome", async () => {
|
||||
const error = transientErrors[1]!.make();
|
||||
const execute = vi.fn<CommandManagedRuntimeRunner["execute"]>().mockRejectedValue(error);
|
||||
await expect(workFolderTransport({ execute }).moveRoot("/old", root)).rejects.toBe(error);
|
||||
expect(execute).toHaveBeenCalledTimes(1);
|
||||
expect(request(execute.mock.calls[0]![0]).operation).toBe("move-root");
|
||||
expect(clock.delay).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
|
@ -159,13 +159,14 @@ describe("sandbox work folder transport with real Node and Git", () => {
|
|||
await expect(fast.write(dir, staging, { ...entry, path: "../private" }, invalid)).rejects.toThrow();
|
||||
expect(invalid.destroyed).toBe(true);
|
||||
});
|
||||
it("does not retry a lost batch response or publish an incomplete source", async () => {
|
||||
it("does not replay a lost batch without a verified missing receipt or publish an incomplete source", async () => {
|
||||
const execute = vi.fn().mockRejectedValue(new Error("socket hang up"));
|
||||
const fast = workFolderTransport({ execute, supportsSingleStreamStdinProgress: true });
|
||||
const entry = { path: "file", kind: "file" as const, byteSize: 3,
|
||||
sha256: createHash("sha256").update("new").digest("hex"), executable: false };
|
||||
await expect(fast.write("/task", "/staging", entry, Readable.from(["new"]))).rejects.toThrow("socket hang up");
|
||||
expect(execute).toHaveBeenCalledTimes(1);
|
||||
const operations = execute.mock.calls.map(([input]) => JSON.parse(Buffer.from(input.args!.at(-1)!, "base64").toString()).operation);
|
||||
expect(operations).toEqual(["batch", "batch-status", "batch-status", "batch-status"]);
|
||||
execute.mockClear();
|
||||
await expect(fast.write("/task", "/staging", entry, Readable.from(["n"]))).rejects.toThrow("size changed");
|
||||
expect(execute).not.toHaveBeenCalled();
|
||||
|
|
|
|||
|
|
@ -3,7 +3,7 @@
|
|||
import fs from "node:fs";
|
||||
import path from "node:path";
|
||||
import os from "node:os";
|
||||
import { createHash } from "node:crypto";
|
||||
import { createHash, createHmac, randomUUID, timingSafeEqual } from "node:crypto";
|
||||
import { execFileSync } from "node:child_process";
|
||||
|
||||
let input = JSON.parse(Buffer.from(process.argv[1], "base64").toString("utf8"));
|
||||
|
|
@ -36,7 +36,7 @@ function withParent(target, create, callback) {
|
|||
fs.closeSync(fd); fd = child;
|
||||
current = path.join(current, segment);
|
||||
}
|
||||
return callback(process.platform === "linux" ? `/proc/self/fd/${fd}/${name}` : path.join(current, name));
|
||||
return callback(process.platform === "linux" ? `/proc/self/fd/${fd}/${name}` : path.join(current, name), fd);
|
||||
} finally { fs.closeSync(fd); }
|
||||
}
|
||||
function checked(target, directory = false) {
|
||||
|
|
@ -207,6 +207,101 @@ if (input.operation === "home") {
|
|||
return result;
|
||||
}
|
||||
|
||||
// Receipts recover uncertain bulk responses, not arbitrary filesystem mutations.
|
||||
// The key detects forged preexisting receipts; it is not an isolation boundary
|
||||
// against the sandbox OS user, who may inspect this process's argv.
|
||||
const RECEIPT_LIMIT = 4096;
|
||||
function batchReceipt(request) {
|
||||
if (!/^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$/.test(request.batchId ?? "")
|
||||
|| !/^[a-f0-9]{64}$/.test(request.batchSha256 ?? "")
|
||||
|| !/^[a-f0-9]{64}$/.test(request.batchReceiptKey ?? "")) throw new Error("invalid_batch_identity");
|
||||
for (const root of [request.root, request.stagingRoot]) {
|
||||
if (typeof root !== "string" || !path.isAbsolute(root) || path.resolve(root) !== root
|
||||
|| /[\x00-\x1f\x7f]/.test(root)) throw new Error("invalid_batch_root");
|
||||
const fd = checked(root, true); fs.closeSync(fd);
|
||||
}
|
||||
const directory = path.join(request.stagingRoot, ".batch-receipts");
|
||||
const target = path.join(directory, request.batchId);
|
||||
const identity = { version: 1, batchId: request.batchId, root: request.root,
|
||||
stagingRoot: request.stagingRoot, batchSha256: request.batchSha256 };
|
||||
const sign = (value) => createHmac("sha256", Buffer.from(request.batchReceiptKey, "hex"))
|
||||
.update(JSON.stringify(value)).digest("hex");
|
||||
function read() {
|
||||
let fd;
|
||||
try { fd = checked(target); } catch (error) { if (error.code === "ENOENT") return { state: "missing" }; throw error; }
|
||||
try {
|
||||
const size = fs.fstatSync(fd).size;
|
||||
if (size === 0) return { state: "running" }; // Exclusive claim, before atomic signed publication.
|
||||
if (size > RECEIPT_LIMIT) throw new Error("invalid_batch_receipt");
|
||||
const buffer = Buffer.alloc(RECEIPT_LIMIT + 1);
|
||||
const count = fs.readSync(fd, buffer, 0, buffer.length, 0);
|
||||
if (count !== size) throw new Error("invalid_batch_receipt");
|
||||
let receipt;
|
||||
try { receipt = JSON.parse(buffer.subarray(0, count).toString("utf8")); }
|
||||
catch { throw new Error("invalid_batch_receipt"); }
|
||||
if (!receipt || typeof receipt !== "object" || Array.isArray(receipt)) throw new Error("invalid_batch_receipt");
|
||||
for (const [key, value] of Object.entries(identity)) {
|
||||
if (receipt[key] !== value) throw new Error("batch_receipt_identity_mismatch");
|
||||
}
|
||||
if (!["running", "completed", "failed"].includes(receipt.state)) throw new Error("invalid_batch_receipt");
|
||||
const payload = { ...identity, state: receipt.state };
|
||||
if (receipt.state === "completed") {
|
||||
if (!Number.isSafeInteger(receipt.completed) || receipt.completed < 0 || receipt.completed > 512) throw new Error("invalid_batch_receipt");
|
||||
payload.completed = receipt.completed;
|
||||
} else if (receipt.state === "failed") {
|
||||
if (receipt.error !== "batch_execution_failed") throw new Error("invalid_batch_receipt");
|
||||
payload.error = receipt.error;
|
||||
}
|
||||
const keys = [...Object.keys(payload), "signature"].sort();
|
||||
if (JSON.stringify(Object.keys(receipt).sort()) !== JSON.stringify(keys)
|
||||
|| !/^[a-f0-9]{64}$/.test(receipt.signature ?? "")
|
||||
|| !timingSafeEqual(Buffer.from(receipt.signature, "hex"), Buffer.from(sign(payload), "hex"))) throw new Error("invalid_batch_receipt");
|
||||
return receipt.state === "completed" ? { state: "completed", completed: receipt.completed }
|
||||
: receipt.state === "failed" ? { state: "failed", error: receipt.error } : { state: "running" };
|
||||
} finally { fs.closeSync(fd); }
|
||||
}
|
||||
function replace(state, completed) {
|
||||
const payload = { ...identity, state, ...(state === "completed" ? { completed }
|
||||
: state === "failed" ? { error: "batch_execution_failed" } : {}) };
|
||||
const bytes = Buffer.from(JSON.stringify({ ...payload, signature: sign(payload) }));
|
||||
if (bytes.length > RECEIPT_LIMIT) throw new Error("invalid_batch_receipt");
|
||||
const temporary = path.join(directory, `${request.batchId}.${randomUUID()}.tmp`);
|
||||
withParent(temporary, false, (from) => {
|
||||
const fd = fs.openSync(from, fs.constants.O_WRONLY | fs.constants.O_CREAT | fs.constants.O_EXCL | fs.constants.O_NOFOLLOW, 0o600);
|
||||
try {
|
||||
if (!fs.fstatSync(fd).isFile() || fs.fstatSync(fd).nlink !== 1) throw new Error("unsupported_file");
|
||||
fs.writeFileSync(fd, bytes); fs.fsyncSync(fd);
|
||||
} finally { fs.closeSync(fd); }
|
||||
try {
|
||||
const sourceFd = checked(temporary); fs.closeSync(sourceFd);
|
||||
const targetFd = checked(target); fs.closeSync(targetFd);
|
||||
withParent(target, false, (to, parentFd) => { fs.renameSync(from, to); fs.fsyncSync(parentFd); });
|
||||
} finally {
|
||||
try { fs.unlinkSync(from); } catch (error) { if (error.code !== "ENOENT") throw error; }
|
||||
}
|
||||
});
|
||||
}
|
||||
function claim() {
|
||||
ensureDirectory(directory);
|
||||
try {
|
||||
withParent(target, false, (anchored, parentFd) => {
|
||||
const fd = fs.openSync(anchored, fs.constants.O_WRONLY | fs.constants.O_CREAT | fs.constants.O_EXCL | fs.constants.O_NOFOLLOW, 0o600);
|
||||
try { fs.fsyncSync(fd); fs.fsyncSync(parentFd); } finally { fs.closeSync(fd); }
|
||||
});
|
||||
} catch (error) { if (error.code === "EEXIST") return false; throw error; }
|
||||
replace("running");
|
||||
return true;
|
||||
}
|
||||
function confined(operation) {
|
||||
if (Object.hasOwn(operation, "root") || Object.hasOwn(operation, "stagingRoot") || Object.hasOwn(operation, "source")) throw new Error("invalid_batch_operation");
|
||||
const root = operation.operation === "write" ? request.stagingRoot : request.root;
|
||||
const targets = [path.join(root, safeRelative(operation.path))];
|
||||
if (operation.operation === "publish") targets.push(path.join(request.stagingRoot, safeRelative(operation.stagingPath)));
|
||||
if (targets.some(value => value === directory || value.startsWith(`${directory}/`))) throw new Error("reserved_batch_receipt_path");
|
||||
}
|
||||
return { read, replace, claim, confined };
|
||||
}
|
||||
|
||||
const request = input;
|
||||
if (request.operation === "read-batch") {
|
||||
if (!Array.isArray(request.entries) || request.entries.length > 64) throw new Error("invalid_read_batch");
|
||||
|
|
@ -220,9 +315,10 @@ if (request.operation === "read-batch") {
|
|||
const results = request.entries.map((entry) => execute({ operation: "read", root: request.root,
|
||||
path: entry.path, offset: 0, length: Math.max(1, entry.byteSize) }));
|
||||
process.stdout.write(JSON.stringify(results));
|
||||
} else if (request.operation === "batch-status") {
|
||||
process.stdout.write(JSON.stringify(batchReceipt(request).read()));
|
||||
} else if (request.operation === "batch") {
|
||||
// The provider transports stdin as one bounded upload. The roots stay in the
|
||||
// host-authored argv envelope, so batch contents cannot redirect operations.
|
||||
// Stdin is bounded and hashed exactly, before parsing or claiming execution.
|
||||
const chunks = [];
|
||||
let size = 0;
|
||||
const buffer = Buffer.alloc(64 * 1024);
|
||||
|
|
@ -232,15 +328,38 @@ if (request.operation === "read-batch") {
|
|||
if (size > 8 * 1024 * 1024) throw new Error("batch_too_large");
|
||||
chunks.push(Buffer.from(buffer.subarray(0, count)));
|
||||
}
|
||||
const operations = JSON.parse(Buffer.concat(chunks).toString("utf8"));
|
||||
const body = Buffer.concat(chunks);
|
||||
if (createHash("sha256").update(body).digest("hex") !== request.batchSha256) throw new Error("batch_body_hash_mismatch");
|
||||
const receipt = batchReceipt(request);
|
||||
const operations = JSON.parse(body.toString("utf8"));
|
||||
if (!Array.isArray(operations) || operations.length > 512) throw new Error("invalid_batch");
|
||||
for (const operation of operations) {
|
||||
if (!operation || !["write", "publish", "mkdir"].includes(operation.operation)) throw new Error("invalid_batch_operation");
|
||||
// Never accept roots, source roots or filesystem commands from the body.
|
||||
execute({ ...operation, root: operation.operation === "write" ? request.stagingRoot : request.root,
|
||||
stagingRoot: request.stagingRoot });
|
||||
receipt.confined(operation);
|
||||
}
|
||||
if (!receipt.claim()) {
|
||||
const status = receipt.read();
|
||||
if (status.state === "completed") process.stdout.write(JSON.stringify({ completed: status.completed }));
|
||||
else if (status.state === "running") process.stdout.write(JSON.stringify({ pending: true }));
|
||||
else throw new Error(status.state === "failed" ? "batch_execution_failed" : "batch_receipt_disappeared");
|
||||
} else {
|
||||
try {
|
||||
for (const operation of operations) {
|
||||
// Roots always come from the host envelope, never the batch body.
|
||||
execute({ ...operation, root: operation.operation === "write" ? request.stagingRoot : request.root,
|
||||
stagingRoot: request.stagingRoot });
|
||||
}
|
||||
receipt.replace("completed", operations.length);
|
||||
} catch (error) {
|
||||
receipt.replace("failed");
|
||||
// Known helper codes contain no paths, file bytes or credentials. Native
|
||||
// filesystem error messages may contain paths, so keep those private.
|
||||
const safeErrors = ["unsafe_path", "invalid_root", "symlink_not_allowed", "unsupported_file",
|
||||
"hardlink_not_allowed", "chunk_too_large", "invalid_chunk_offset", "content_changed_during_transfer"];
|
||||
throw new Error(safeErrors.includes(error.message) ? error.message : "batch_execution_failed");
|
||||
}
|
||||
process.stdout.write(JSON.stringify({ completed: operations.length }));
|
||||
}
|
||||
process.stdout.write(JSON.stringify({ completed: operations.length }));
|
||||
} else {
|
||||
process.stdout.write(JSON.stringify(execute(request)));
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
import { readFile } from "node:fs/promises";
|
||||
import { Readable } from "node:stream";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { createHash, randomBytes, randomUUID } from "node:crypto";
|
||||
import { setTimeout as delay } from "node:timers/promises";
|
||||
import path from "node:path";
|
||||
import { z } from "zod";
|
||||
|
|
@ -19,11 +19,17 @@ const BATCH_BYTES = 4 * 1024 * 1024;
|
|||
const BATCH_OPERATIONS = 256;
|
||||
export type WorkFileTransfer = { entry: WorkTreeEntry; body?: Readable };
|
||||
|
||||
function transientReadFailure(error: unknown) {
|
||||
function transientTransportFailure(error: unknown) {
|
||||
if (!(error instanceof Error)) return false;
|
||||
const code = (error as Error & { code?: string }).code;
|
||||
return ["ECONNRESET", "EPIPE", "EAI_AGAIN", "ECONNABORTED"].includes(code ?? "")
|
||||
|| error.message === "socket hang up";
|
||||
const detail = error as Error & { code?: unknown; status?: unknown; statusCode?: unknown; response?: { status?: unknown } };
|
||||
const statuses = [502, 503, 504];
|
||||
return ["ECONNRESET", "EPIPE", "EAI_AGAIN", "ECONNABORTED"].includes(String(detail.code ?? ""))
|
||||
|| error.message === "socket hang up"
|
||||
|| [detail.status, detail.statusCode, detail.response?.status].some((status) => statuses.includes(status as number))
|
||||
// The plugin RPC preserves this SDK error's message but not its response
|
||||
// metadata. Do not classify arbitrary script output containing "502".
|
||||
|| (error.name === "JsonRpcCallError" && detail.code === -32002
|
||||
&& /^Request failed with status code (502|503|504)(?:: Sandbox command requested here)?$/.test(error.message));
|
||||
}
|
||||
|
||||
export function workFolderTransport(runner: CommandManagedRuntimeRunner) {
|
||||
|
|
@ -31,13 +37,13 @@ export function workFolderTransport(runner: CommandManagedRuntimeRunner) {
|
|||
// runners explicitly advertise streaming stdin. Other providers retain the
|
||||
// small-argv transport without assuming additional capabilities.
|
||||
const bulkStdin = Boolean(runner.syncIn && runner.syncOut) || runner.supportsSingleStreamStdinProgress === true;
|
||||
async function command(input: Record<string, unknown>, stdin?: string): Promise<unknown> {
|
||||
async function command(input: Record<string, unknown>, stdin?: string, deadline = Date.now() + 120_000): Promise<unknown> {
|
||||
source ??= readFile(new URL("./scripts/work-folder-io.mjs", import.meta.url), "utf8");
|
||||
const args = ["--input-type=module", "-e", await source, Buffer.from(JSON.stringify(input)).toString("base64")];
|
||||
const readOnly = ["home", "scan", "read", "read-batch"].includes(String(input.operation));
|
||||
const deadline = Date.now() + 120_000;
|
||||
const readOnly = ["home", "scan", "read", "read-batch", "batch-status"].includes(String(input.operation));
|
||||
let result;
|
||||
for (let attempt = 0; ; attempt++) {
|
||||
if (Date.now() >= deadline) throw new Error("Work folder transfer deadline exceeded");
|
||||
try {
|
||||
result = await runner.execute({ command: "node", args, ...(stdin === undefined ? {} : { stdin }), bypassSession: true,
|
||||
timeoutMs: Math.max(1, deadline - Date.now()) });
|
||||
|
|
@ -46,13 +52,53 @@ export function workFolderTransport(runner: CommandManagedRuntimeRunner) {
|
|||
// A lost read response is safe to repeat. Staging writes, publishes and
|
||||
// moves may already have happened, so never replay them here.
|
||||
const waitMs = 250 * (attempt + 1);
|
||||
if (!readOnly || attempt >= 2 || !transientReadFailure(error) || Date.now() + waitMs >= deadline) throw error;
|
||||
if (!readOnly || attempt >= 2 || !transientTransportFailure(error) || Date.now() + waitMs >= deadline) throw error;
|
||||
await delay(waitMs);
|
||||
}
|
||||
}
|
||||
if (result.exitCode !== 0 || result.timedOut) throw new Error(`Work folder ${String(input.operation)} failed: ${result.stderr.slice(0, 1500)}`);
|
||||
return JSON.parse(result.stdout);
|
||||
}
|
||||
async function writeBatch(root: string, stagingRoot: string, body: string) {
|
||||
const deadline = Date.now() + 120_000;
|
||||
const identity = { root, stagingRoot, batchId: randomUUID(),
|
||||
batchSha256: createHash("sha256").update(body).digest("hex"), batchReceiptKey: randomBytes(32).toString("hex") };
|
||||
const completedSchema = z.object({ completed: z.number().int().nonnegative().max(512) });
|
||||
const resultSchema = z.union([completedSchema, z.object({ pending: z.literal(true) })]);
|
||||
const statusSchema = z.discriminatedUnion("state", [
|
||||
z.object({ state: z.literal("missing") }), z.object({ state: z.literal("running") }),
|
||||
completedSchema.extend({ state: z.literal("completed") }),
|
||||
z.object({ state: z.literal("failed"), error: z.string().max(500) }),
|
||||
]);
|
||||
let attempts = 0;
|
||||
let pending = false;
|
||||
let lastError: unknown;
|
||||
while (Date.now() < deadline) {
|
||||
if (!pending) {
|
||||
attempts++;
|
||||
try {
|
||||
const result = resultSchema.parse(await command({ operation: "batch", ...identity }, body, deadline));
|
||||
if ("completed" in result) return result;
|
||||
pending = true;
|
||||
} catch (error) {
|
||||
if (!transientTransportFailure(error)) throw error;
|
||||
lastError = error;
|
||||
}
|
||||
}
|
||||
// An upstream error may have lost only the response. First inspect the
|
||||
// sandbox receipt. A claimed batch is never replayed; it must complete
|
||||
// or remain visibly recoverable when this bounded wait expires.
|
||||
const status = statusSchema.parse(await command({ operation: "batch-status", ...identity }, undefined, deadline));
|
||||
if (status.state === "completed") return { completed: status.completed };
|
||||
if (status.state === "failed") throw new Error(`Work folder batch failed: ${status.error}`);
|
||||
pending = status.state === "running";
|
||||
if (!pending && attempts >= 3) throw lastError ?? new Error("Work folder batch receipt is missing");
|
||||
const waitMs = pending ? 500 : 250 * attempts;
|
||||
if (Date.now() + waitMs >= deadline) break;
|
||||
await delay(waitMs);
|
||||
}
|
||||
throw new Error("Work folder batch outcome is uncertain; retaining sandbox for recovery");
|
||||
}
|
||||
async function home() {
|
||||
const result = z.object({ home: z.string().startsWith("/") }).parse(await command({ operation: "home" }));
|
||||
return result.home;
|
||||
|
|
@ -118,7 +164,7 @@ export function workFolderTransport(runner: CommandManagedRuntimeRunner) {
|
|||
const body = JSON.stringify(operations);
|
||||
if (Buffer.byteLength(body) > 8 * 1024 * 1024) throw new Error("Work folder batch exceeds transfer limit");
|
||||
if (publishedEntries.length) await beforePublish?.(publishedEntries);
|
||||
const result = z.object({ completed: z.number().int() }).parse(await command({ operation: "batch", root, stagingRoot }, body));
|
||||
const result = await writeBatch(root, stagingRoot, body);
|
||||
if (result.completed !== operations.length) throw new Error("Work folder batch did not complete");
|
||||
operations = []; bufferedBytes = 0; publishedEntries = [];
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue