mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-31 14:16:39 +01:00
fix(runtime): stream workspace snapshots through disk manifests
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Paperclip authored and Dotta committed
1 parent
637b6615d5
commit
38441fb2e0
17 files changed
+1172
-482
No files matched your search
+11
-3
@@ -702,9 +702,16 @@ When effective run config changes, Paperclip may intentionally skip a saved adap
|
||||
|
||||
## Workspace Git Scan Protection
|
||||
|
||||
Paperclip applies one process-wide scheduler to expensive host-side workspace Git enumeration, including changed-file browsing, runtime/finalization cleanliness guards, and adapter sandbox-sync snapshots. The scheduler defaults to two active scans and a bounded queue of 32. Identical scans of the same canonical worktree share one subprocess, while successful changed-file listings are cached for 10 seconds. Correctness-sensitive runtime guards bypass the result cache.
|
||||
Paperclip applies one process-wide scheduler to expensive host-side workspace Git enumeration, including changed-file browsing, runtime/finalization cleanliness guards, and adapter sandbox-sync snapshots. The scheduler defaults to two active scans and a bounded queue of 32. Identical buffered scans of the same canonical worktree share one subprocess, while successful changed-file listings are cached for 10 seconds. Streaming snapshot scans have caller-owned sinks, so they use separate jobs in the same queue and are never cached or coalesced. Correctness-sensitive runtime guards bypass the result cache.
|
||||
|
||||
Workspace snapshots list ignored paths with `git ls-files --others --ignored --exclude-standard --directory -z` so ignored directory contents do not require a full status walk. The untracked-file snapshot has a 32 MiB output bound so generated trees with thousands of long filenames can be staged. It still records explicit file paths; files created after the scan do not enter that overlay. Other scan bounds stay unchanged. Snapshot failures retain their typed cause instead of becoming a non-Git-folder result. During pre-provider setup, scan timeouts and queue saturation use the existing two automatic failure retries with a 30-second delay. Cancellation, output limits, and other Git errors stop with specific recovery guidance. See `doc/execution-semantics.md` for the ownership and retry-budget contract.
|
||||
Workspace snapshots list ignored paths with `git ls-files --others --ignored --exclude-standard --directory -z` so ignored directory contents do not require a full status walk. Changed, untracked, deleted, and ignored filename lists stream into private SQLite manifests. There is no total filename-list byte limit. The parser and each sink chunk are limited to 64 KiB, and each SQLite connection uses a 1 MiB page cache with disk-backed temporary storage. Records must be complete NUL-delimited UTF-8 paths. Invalid, oversized, or incomplete records fail the scan. Explicit file selection excludes files created after the scan. Buffered browser/guard and referenced-source scan bounds stay unchanged. Snapshot failures retain their typed cause instead of becoming a non-Git-folder result. During pre-provider setup, scan timeouts and queue saturation use the existing two automatic failure retries with a 30-second delay. Cancellation, output limits, and other Git errors stop with specific recovery guidance. See `doc/execution-semantics.md` for the ownership and retry-budget contract.
|
||||
|
||||
|
||||
Snapshot scans default to a 30-minute execution deadline, including disk backpressure, rather than the short interactive scan deadline. Set `PAPERCLIP_WORKSPACE_GIT_SNAPSHOT_TIMEOUT_MS` to allow slower storage or larger trees (1 second to 24 hours). Disk capacity, filesystem path limits, Git's own resource use, and this execution budget remain practical limits. The scheduler keeps the active slot until the child, pipes, and pending sink write finish. Snapshot failures cancel sibling producers and settle all started producers before removing temporary storage.
|
||||
|
||||
Workspace baselines, nested-repository subsets, and merge lookups also use disk-backed manifests. Local archive input uses NUL list files. Deletion replay uploads a NUL manifest and runs bounded argument batches with ancestor symlink checks. Git ignored paths are literal indexed exclusions for staging, baseline capture, and the host merge. Restore transport uses the fixed/configured archive exclusions; it can transfer a remotely created ignored file, but the host merge excludes it. The v2 native recovery descriptor contains manifest references, not file arrays. Recovery checks controller-owned paths and content digests before reading records. Existing v1 descriptors remain readable. Transient manifests are removed on preparation failure and restore completion; native manifests remain until the durable recovery lifecycle is cleaned up.
|
||||
|
||||
These bounds apply to application filename storage, not total process memory. Repository metadata scales with the number of managed repositories; directory iteration retains bounded buffers per nesting level. Git, tar, and provider SDKs have their own memory use. Some provider transfer APIs still return entire archive buffers. Legacy v1 recovery and the separately bounded referenced-source ignore resolver retain their older arrays. This change does not claim constant memory for those paths.
|
||||
|
||||
Workspace preparation resolves an existing root symlink before reading the snapshot and uses that resolved directory for the rest of the operation. Overlay staging checks the captured root identity and each selected path's ancestors before and after copying. A replaced root or a symlink in an ancestor directory stops staging before upload. A missing source file can be skipped; other source inspection errors stop staging. Selected symlink entries remain symlinks.
|
||||
|
||||
@@ -716,7 +723,8 @@ Environment overrides:
|
||||
|
||||
- `PAPERCLIP_WORKSPACE_GIT_SCAN_CONCURRENCY` (default `2`, range `1`–`16`)
|
||||
- `PAPERCLIP_WORKSPACE_GIT_SCAN_QUEUE_CAPACITY` (default `32`, range `0`–`1024`)
|
||||
- `PAPERCLIP_WORKSPACE_GIT_SCAN_TIMEOUT_MS` (default `8000`, range `100`–`120000`)
|
||||
- `PAPERCLIP_WORKSPACE_GIT_SCAN_TIMEOUT_MS` (default `8000`, range `100`–`120000`, buffered scans)
|
||||
- `PAPERCLIP_WORKSPACE_GIT_SNAPSHOT_TIMEOUT_MS` (default `1800000`, range `1000`–`86400000`, streaming snapshots)
|
||||
- `PAPERCLIP_WORKSPACE_GIT_SCAN_CACHE_TTL_MS` (default `10000`, range `0`–`60000`)
|
||||
|
||||
Structured `workspace_git_scan` logs expose the operation name, a non-reversible workspace-path hash, queue and execution durations, active/queued counts, cache and single-flight use, and terminal outcome. Saturation and timeout warnings are rate-limited so an overload does not create a second logging storm.
|
||||
|
||||
@@ -267,6 +267,7 @@ export interface PreparedAdapterExecutionTargetRuntime {
|
||||
baseline: DirectorySnapshot;
|
||||
gitSnapshot: GitWorkspaceSnapshot | null;
|
||||
} | null;
|
||||
cleanupWorkspaceSnapshot?(): Promise<void>;
|
||||
restoreWorkspace(onProgress?: RuntimeProgressSink): Promise<void>;
|
||||
}
|
||||
|
||||
@@ -1522,6 +1523,7 @@ export async function prepareAdapterExecutionTargetRuntime(input: {
|
||||
additionalSourceDirs: prepared.additionalSourceDirs,
|
||||
additionalSourceFailures: prepared.additionalSourceFailures,
|
||||
workspaceSyncSnapshot: prepared.workspaceSyncSnapshot,
|
||||
cleanupWorkspaceSnapshot: prepared.cleanupWorkspaceSnapshot,
|
||||
restoreWorkspace: prepared.restoreWorkspace,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -3,6 +3,8 @@ import { lstat, mkdir, mkdtemp, readFile, readlink, rename, rm, stat, symlink, w
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { promisify } from "node:util";
|
||||
import { workspacePaths } from "./workspace-manifest.js";
|
||||
import { runWorkspaceGitProcess } from "./workspace-git-stream.js";
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
|
||||
import {
|
||||
@@ -13,7 +15,10 @@ import {
|
||||
fetchGitBundleIntoLocalRef,
|
||||
integrateImportedGitHead,
|
||||
isMissingGitPrerequisiteError,
|
||||
readGitWorkspaceSnapshot,
|
||||
readGitWorkspaceSnapshot as readRawSnapshot,
|
||||
disposeGitWorkspaceSnapshot,
|
||||
type ExpensiveWorkspaceGitInput,
|
||||
type GitWorkspaceSnapshot,
|
||||
ReferencedSourceIgnoreScanLimitExceededError,
|
||||
readReferencedSourceGitIgnoredPaths,
|
||||
REFERENCED_SOURCE_IGNORE_MAX_ENTRY_COUNT,
|
||||
@@ -26,6 +31,21 @@ import {
|
||||
|
||||
const execFile = promisify(execFileCallback);
|
||||
|
||||
const snapshots: GitWorkspaceSnapshot[] = [];
|
||||
async function readGitWorkspaceSnapshot(...args: Parameters<typeof readRawSnapshot>) {
|
||||
const snapshot = await readRawSnapshot(...args);
|
||||
if (!snapshot) return null;
|
||||
snapshots.push(snapshot);
|
||||
// Small fixture assertions use arrays; production consumes only iterators.
|
||||
return { ...snapshot, overlayPaths: [...workspacePaths(snapshot.overlayPaths)].sort((a,b)=>a.localeCompare(b)),
|
||||
deletedPaths: [...workspacePaths(snapshot.deletedPaths)], ignoredPaths: [...workspacePaths(snapshot.ignoredPaths)] };
|
||||
}
|
||||
function executeScan(input: ExpensiveWorkspaceGitInput) {
|
||||
if (!input.onStdout) return runLocalGit(input.localDir, [...input.args], input);
|
||||
return runWorkspaceGitProcess({ cwd: input.localDir, args: input.args, timeoutMs: input.timeout,
|
||||
maxStdoutBytes: input.maxBuffer, maxStderrBytes: input.maxBuffer, onStdout: input.onStdout, signal: input.signal, env: input.env });
|
||||
}
|
||||
|
||||
async function git(cwd: string, args: string[]): Promise<string> {
|
||||
return (await runLocalGit(cwd, args)).stdout.trim();
|
||||
}
|
||||
@@ -35,6 +55,7 @@ describe("git workspace sync", () => {
|
||||
|
||||
afterEach(async () => {
|
||||
setExpensiveWorkspaceGitExecutor(null);
|
||||
for (const snapshot of snapshots.splice(0)) await disposeGitWorkspaceSnapshot(snapshot);
|
||||
while (cleanupDirs.length > 0) {
|
||||
const dir = cleanupDirs.pop();
|
||||
if (!dir) continue;
|
||||
@@ -50,10 +71,7 @@ describe("git workspace sync", () => {
|
||||
const operations: string[] = [];
|
||||
setExpensiveWorkspaceGitExecutor(async (input) => {
|
||||
operations.push(input.operation);
|
||||
return await runLocalGit(input.localDir, [...input.args], {
|
||||
timeout: input.timeout,
|
||||
maxBuffer: input.maxBuffer,
|
||||
});
|
||||
return executeScan(input);
|
||||
});
|
||||
|
||||
const snapshot = await readGitWorkspaceSnapshot(repo);
|
||||
@@ -112,7 +130,7 @@ describe("git workspace sync", () => {
|
||||
const failure = Object.assign(new Error("Git enumeration failed"), { code });
|
||||
setExpensiveWorkspaceGitExecutor(async (input) => {
|
||||
if (input.operation === "adapter_sync.ignored_files") throw failure;
|
||||
return runLocalGit(input.localDir, [...input.args]);
|
||||
return executeScan(input);
|
||||
});
|
||||
await expect(readGitWorkspaceSnapshot(repo, false)).rejects.toBe(failure);
|
||||
});
|
||||
@@ -128,7 +146,7 @@ describe("git workspace sync", () => {
|
||||
let ignoredArgs: readonly string[] = [];
|
||||
setExpensiveWorkspaceGitExecutor(async (input) => {
|
||||
if (input.operation === "adapter_sync.ignored_files") ignoredArgs = input.args;
|
||||
return runLocalGit(input.localDir, [...input.args]);
|
||||
return executeScan(input);
|
||||
});
|
||||
expect((await readGitWorkspaceSnapshot(repo))?.ignoredPaths).toEqual(["dependencies", "token.secret"]);
|
||||
expect(ignoredArgs).toEqual(["ls-files", "--others", "--ignored", "--exclude-standard", "--directory", "-z"]);
|
||||
@@ -148,10 +166,7 @@ describe("git workspace sync", () => {
|
||||
maxBuffer: 2 * 1024 * 1024,
|
||||
});
|
||||
expect(Buffer.byteLength(raw.stdout)).toBeGreaterThan(1024 * 1024);
|
||||
setExpensiveWorkspaceGitExecutor((input) => runLocalGit(input.localDir, [...input.args], {
|
||||
timeout: input.timeout,
|
||||
maxBuffer: input.maxBuffer,
|
||||
}));
|
||||
setExpensiveWorkspaceGitExecutor(executeScan);
|
||||
|
||||
const snapshot = await readGitWorkspaceSnapshot(repo);
|
||||
expect(snapshot?.overlayPaths).toEqual(
|
||||
@@ -180,9 +195,14 @@ describe("git workspace sync", () => {
|
||||
await rename(generatedDir, path.join(deepParent, "storybook-output"));
|
||||
expect(Buffer.byteLength(largerRaw.stdout) + 40_000 * (path.relative(repo, deepParent).length + 1))
|
||||
.toBeGreaterThan(32 * 1024 * 1024);
|
||||
await expect(readGitWorkspaceSnapshot(repo)).rejects.toMatchObject({
|
||||
code: "ERR_CHILD_PROCESS_STDIO_MAXBUFFER",
|
||||
});
|
||||
const streamed = await readRawSnapshot(repo);
|
||||
snapshots.push(streamed!);
|
||||
let count = 0;
|
||||
let bytes = 0;
|
||||
for (const relative of workspacePaths(streamed!.overlayPaths)) { count++; bytes += Buffer.byteLength(relative) + 1; }
|
||||
expect(count).toBe(40_000);
|
||||
expect(bytes).toBeGreaterThan(32 * 1024 * 1024);
|
||||
expect(JSON.stringify(streamed).length).toBeLessThan(4096);
|
||||
}, 60_000);
|
||||
|
||||
async function createRepo(rootDir: string): Promise<string> {
|
||||
|
||||
@@ -3,6 +3,8 @@ import { execFile } from "node:child_process";
|
||||
import { promises as fs } from "node:fs";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { createWorkspaceManifest, workspacePaths, WorkspaceNulParser, type WorkspacePaths } from "./workspace-manifest.js";
|
||||
import { runWorkspaceGitProcess } from "./workspace-git-stream.js";
|
||||
|
||||
export interface GitCommandResult {
|
||||
stdout: string;
|
||||
@@ -12,9 +14,9 @@ export interface GitCommandResult {
|
||||
export interface GitWorkspaceSnapshot {
|
||||
headCommit: string;
|
||||
branchName: string | null;
|
||||
overlayPaths: string[];
|
||||
deletedPaths: string[];
|
||||
ignoredPaths: string[];
|
||||
overlayPaths: WorkspacePaths;
|
||||
deletedPaths: WorkspacePaths;
|
||||
ignoredPaths: WorkspacePaths;
|
||||
/** Managed, editable repositories inside the task workspace. */
|
||||
repositories?: Array<{ path: string; snapshot: GitWorkspaceSnapshot }>;
|
||||
}
|
||||
@@ -27,6 +29,8 @@ export interface ExpensiveWorkspaceGitInput {
|
||||
operation: string;
|
||||
timeout: number;
|
||||
maxBuffer: number;
|
||||
onStdout?: (chunk: Buffer) => Promise<void> | void;
|
||||
signal?: AbortSignal;
|
||||
/**
|
||||
* Optional environment override for the invocation. Absent for the anchor
|
||||
* workspace's own full-tree walks (they inherit the process environment, a
|
||||
@@ -60,7 +64,7 @@ export const WORKSPACE_GIT_SCAN_SATURATED_CODE = "workspace_git_scan_saturated";
|
||||
/**
|
||||
* Lets a host process apply its process-wide admission policy to the adapter
|
||||
* package's full-tree Git walks. Standalone adapter-utils consumers retain the
|
||||
* existing timeout/buffer-bounded fallback.
|
||||
* streaming process fallback for filename scans.
|
||||
*/
|
||||
export function setExpensiveWorkspaceGitExecutor(executor: ExpensiveWorkspaceGitExecutor | null): void {
|
||||
expensiveWorkspaceGitExecutor = executor;
|
||||
@@ -125,7 +129,7 @@ async function runExpensiveWorkspaceGit(
|
||||
localDir: string,
|
||||
args: string[],
|
||||
operation: string,
|
||||
options: { timeout: number; maxBuffer: number; env?: NodeJS.ProcessEnv },
|
||||
options: { timeout: number; maxBuffer: number; env?: NodeJS.ProcessEnv; onStdout?: ExpensiveWorkspaceGitInput["onStdout"]; signal?: AbortSignal },
|
||||
): Promise<GitCommandResult> {
|
||||
if (expensiveWorkspaceGitExecutor) {
|
||||
return await expensiveWorkspaceGitExecutor({
|
||||
@@ -135,30 +139,36 @@ async function runExpensiveWorkspaceGit(
|
||||
timeout: options.timeout,
|
||||
maxBuffer: options.maxBuffer,
|
||||
env: options.env,
|
||||
onStdout: options.onStdout,
|
||||
signal: options.signal,
|
||||
});
|
||||
}
|
||||
if (options.onStdout) return runWorkspaceGitProcess({
|
||||
cwd: localDir, args, timeoutMs: options.timeout, maxStdoutBytes: options.maxBuffer,
|
||||
maxStderrBytes: options.maxBuffer, onStdout: options.onStdout, signal: options.signal, env: options.env,
|
||||
});
|
||||
return await runLocalGit(localDir, args, options);
|
||||
}
|
||||
|
||||
export async function readGitWorkspaceSnapshot(localDir: string, includeRepositories = true): Promise<GitWorkspaceSnapshot | null> {
|
||||
const ownedSnapshots = new WeakMap<GitWorkspaceSnapshot, string>();
|
||||
|
||||
export async function disposeGitWorkspaceSnapshot(snapshot: GitWorkspaceSnapshot | null): Promise<void> {
|
||||
if (!snapshot) return;
|
||||
const ownedDirectory = ownedSnapshots.get(snapshot);
|
||||
if (!ownedDirectory) return;
|
||||
ownedSnapshots.delete(snapshot);
|
||||
for (const repository of snapshot.repositories ?? []) await disposeGitWorkspaceSnapshot(repository.snapshot);
|
||||
await fs.rm(ownedDirectory, { recursive: true, force: true });
|
||||
}
|
||||
|
||||
/** Snapshot deadlines include disk backpressure. Operators can allow up to 24h. */
|
||||
export function workspaceSnapshotTimeoutMs(): number {
|
||||
const configured = Number(process.env.PAPERCLIP_WORKSPACE_GIT_SNAPSHOT_TIMEOUT_MS);
|
||||
return Number.isFinite(configured) && configured >= 1000 ? Math.min(configured, 86_400_000) : 30 * 60_000;
|
||||
}
|
||||
|
||||
export async function readGitWorkspaceSnapshot(localDir: string, includeRepositories = true, options: { signal?: AbortSignal } = {}): Promise<GitWorkspaceSnapshot | null> {
|
||||
const repositories: NonNullable<GitWorkspaceSnapshot["repositories"]> = [];
|
||||
if (includeRepositories) {
|
||||
const root = path.join(localDir, PROJECT_REPOSITORIES_DIR);
|
||||
const rootStat = await fs.lstat(root).catch((error: NodeJS.ErrnoException) => {
|
||||
if (error.code === "ENOENT") return null;
|
||||
throw error;
|
||||
});
|
||||
if (rootStat) {
|
||||
if (!rootStat.isDirectory() || rootStat.isSymbolicLink()) throw new Error("Invalid project repositories directory");
|
||||
for (const entry of (await fs.readdir(root, { withFileTypes: true })).sort((a, b) => a.name.localeCompare(b.name))) {
|
||||
if (!entry.isDirectory() || !/^[a-zA-Z0-9_-]+$/.test(entry.name)) throw new Error("Invalid project repository directory");
|
||||
const relative = `${PROJECT_REPOSITORIES_DIR}/${entry.name}`;
|
||||
const snapshot = await readGitWorkspaceSnapshot(path.join(localDir, relative), false);
|
||||
if (!snapshot) throw new Error(`Project repository is not a Git checkout: ${relative}`);
|
||||
repositories.push({ path: relative, snapshot });
|
||||
}
|
||||
}
|
||||
}
|
||||
// Only repository discovery may report an ordinary directory. A failed
|
||||
// snapshot of a confirmed repository must never fall back to directory sync.
|
||||
let insideWorkTree: GitCommandResult;
|
||||
@@ -188,65 +198,86 @@ export async function readGitWorkspaceSnapshot(localDir: string, includeReposito
|
||||
]);
|
||||
if (workspacePath !== repositoryPath) return null;
|
||||
|
||||
const [headCommitResult, branchResult, overlayDiffResult, untrackedResult, deletedResult, ignoredResult] = await Promise.all([
|
||||
runLocalGit(localDir, ["rev-parse", "HEAD"], {
|
||||
timeout: 10_000,
|
||||
maxBuffer: 16 * 1024,
|
||||
}),
|
||||
runLocalGit(localDir, ["rev-parse", "--abbrev-ref", "HEAD"], {
|
||||
timeout: 10_000,
|
||||
maxBuffer: 16 * 1024,
|
||||
}),
|
||||
runExpensiveWorkspaceGit(localDir, ["diff", "--name-only", "-z", "--diff-filter=ACMRTUXB", "HEAD", "--"], "adapter_sync.overlay_diff", {
|
||||
timeout: 10_000,
|
||||
maxBuffer: 1024 * 1024,
|
||||
}),
|
||||
// A generated output tree can exceed 1 MiB of filenames with only a few
|
||||
// thousand files. Keep the explicit file snapshot (and a finite bound):
|
||||
// collapsing directories would let later files enter the staging copy.
|
||||
runExpensiveWorkspaceGit(localDir, ["ls-files", "--others", "--exclude-standard", "-z"], "adapter_sync.untracked_files", {
|
||||
timeout: 10_000,
|
||||
maxBuffer: 32 * 1024 * 1024,
|
||||
}),
|
||||
runExpensiveWorkspaceGit(localDir, ["diff", "--name-only", "-z", "--diff-filter=D", "HEAD", "--"], "adapter_sync.deleted_files", {
|
||||
timeout: 10_000,
|
||||
maxBuffer: 256 * 1024,
|
||||
}),
|
||||
// Collapse ignored directories instead of walking their contents, and
|
||||
// avoid producing unrelated tracked/untracked status records.
|
||||
runExpensiveWorkspaceGit(localDir, ["ls-files", "--others", "--ignored", "--exclude-standard", "--directory", "-z"], "adapter_sync.ignored_files", {
|
||||
timeout: 10_000,
|
||||
maxBuffer: 1024 * 1024,
|
||||
}),
|
||||
]);
|
||||
const writer = await createWorkspaceManifest();
|
||||
const controller = new AbortController();
|
||||
const abort = () => controller.abort();
|
||||
options.signal?.addEventListener("abort", abort, { once: true });
|
||||
if (options.signal?.aborted) abort();
|
||||
try {
|
||||
if (includeRepositories) {
|
||||
const root = path.join(localDir, PROJECT_REPOSITORIES_DIR);
|
||||
const rootStat = await fs.lstat(root).catch((error: NodeJS.ErrnoException) => {
|
||||
if (error.code === "ENOENT") return null;
|
||||
throw error;
|
||||
});
|
||||
if (rootStat) {
|
||||
if (!rootStat.isDirectory() || rootStat.isSymbolicLink()) throw new Error("Invalid project repositories directory");
|
||||
for await (const entry of await fs.opendir(root)) {
|
||||
if (!entry.isDirectory() || !/^[a-zA-Z0-9_-]+$/.test(entry.name)) throw new Error("Invalid project repository directory");
|
||||
const relative = `${PROJECT_REPOSITORIES_DIR}/${entry.name}`;
|
||||
const snapshot = await readGitWorkspaceSnapshot(path.join(localDir, relative), false, options);
|
||||
if (!snapshot) throw new Error(`Project repository is not a Git checkout: ${relative}`);
|
||||
repositories.push({ path: relative, snapshot });
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const branchName = branchResult.stdout.trim();
|
||||
// `-z` already delimits each record with a NUL byte, so a leading or
|
||||
// trailing space in a record is part of the path itself, not padding to
|
||||
// remove — trimming it would resolve to a path that does not exist. A
|
||||
// length check finds the one genuinely empty record `-z` appends after
|
||||
// the last NUL, without eating a real path's own leading or trailing
|
||||
// whitespace. This applies to all four NUL-delimited outputs below (the
|
||||
// overlay diff, the untracked list, the deleted list, and the ignored
|
||||
// list); `branchName` and `headCommit` come from non-`-z` commands and
|
||||
// keep their own `.trim()` above and below, which is safe.
|
||||
const splitNul = (value: string) => value.split("\0").filter((entry) => entry.length > 0);
|
||||
return {
|
||||
headCommit: headCommitResult.stdout.trim(),
|
||||
branchName: branchName && branchName !== "HEAD" ? branchName : null,
|
||||
overlayPaths: [...new Set([...splitNul(overlayDiffResult.stdout), ...splitNul(untrackedResult.stdout),
|
||||
...repositories.flatMap((repo) => repo.snapshot.overlayPaths.map((entry) => `${repo.path}/${entry}`))])]
|
||||
.sort((left, right) => left.localeCompare(right)),
|
||||
deletedPaths: [...new Set([...splitNul(deletedResult.stdout),
|
||||
...repositories.flatMap((repo) => repo.snapshot.deletedPaths.map((entry) => `${repo.path}/${entry}`))])]
|
||||
.sort((left, right) => left.localeCompare(right)),
|
||||
ignoredPaths: [...splitNul(ignoredResult.stdout)
|
||||
.map((entry) => entry.replace(/\/+$/, ""))
|
||||
.filter((entry) => Boolean(entry) && !(repositories.length > 0 && entry === PROJECT_REPOSITORIES_DIR)),
|
||||
...repositories.flatMap((repo) => repo.snapshot.ignoredPaths.map((entry) => `${repo.path}/${entry}`))]
|
||||
.sort((left, right) => left.localeCompare(right)),
|
||||
...(repositories.length > 0 ? { repositories } : {}),
|
||||
};
|
||||
const scan = async (args: string[], operation: string, category: string) => {
|
||||
const parser = new WorkspaceNulParser((record) => {
|
||||
const relative = category === "ignored" ? record.replace(/\/+$/, "") : record;
|
||||
if (category === "ignored" && repositories.length && relative === PROJECT_REPOSITORIES_DIR) return;
|
||||
// Managed children have their own exact file selections. Never stage a
|
||||
// parent Git directory record that could admit files created later.
|
||||
if (repositories.some((repo) => relative === repo.path || relative.startsWith(`${repo.path}/`))) return;
|
||||
writer.add(category, relative);
|
||||
});
|
||||
const result = await runExpensiveWorkspaceGit(localDir, args, operation, {
|
||||
timeout: workspaceSnapshotTimeoutMs(), maxBuffer: 64 * 1024,
|
||||
signal: controller.signal,
|
||||
onStdout: (chunk) => writer.batch(() => parser.write(chunk)),
|
||||
});
|
||||
if (result.stdout) throw new Error("Workspace Git executor did not stream stdout");
|
||||
try { parser.finish(); }
|
||||
catch (error) { throw Object.assign(error as Error, { code: "workspace_git_scan_failed" }); }
|
||||
};
|
||||
// Cancellation is followed by an all-settled barrier BEFORE storage cleanup.
|
||||
const tasks = [
|
||||
runLocalGit(localDir, ["rev-parse", "HEAD"], { timeout: 10_000, maxBuffer: 16 * 1024 }),
|
||||
runLocalGit(localDir, ["rev-parse", "--abbrev-ref", "HEAD"], { timeout: 10_000, maxBuffer: 16 * 1024 }),
|
||||
scan(["diff", "--name-only", "-z", "--diff-filter=ACMRTUXB", "HEAD", "--"], "adapter_sync.overlay_diff", "overlay"),
|
||||
scan(["ls-files", "--others", "--exclude-standard", "-z"], "adapter_sync.untracked_files", "overlay"),
|
||||
scan(["diff", "--name-only", "-z", "--diff-filter=D", "HEAD", "--"], "adapter_sync.deleted_files", "deleted"),
|
||||
scan(["ls-files", "--others", "--ignored", "--exclude-standard", "--directory", "-z"], "adapter_sync.ignored_files", "ignored"),
|
||||
];
|
||||
let firstError: unknown;
|
||||
const settled = await Promise.allSettled(tasks.map((task) => task.catch((error) => {
|
||||
if (firstError === undefined) firstError = error;
|
||||
controller.abort();
|
||||
throw error;
|
||||
})));
|
||||
if (firstError !== undefined) throw firstError;
|
||||
if (controller.signal.aborted) throw Object.assign(new Error("Workspace snapshot cancelled"), { code: "workspace_git_scan_cancelled" });
|
||||
for (const repo of repositories) {
|
||||
for (const [category, paths] of [["overlay", repo.snapshot.overlayPaths], ["deleted", repo.snapshot.deletedPaths], ["ignored", repo.snapshot.ignoredPaths]] as const) {
|
||||
for (const relative of workspacePaths(paths)) writer.add(category, `${repo.path}/${relative}`);
|
||||
}
|
||||
}
|
||||
const head = (settled[0] as PromiseFulfilledResult<GitCommandResult>).value.stdout.trim();
|
||||
const branch = (settled[1] as PromiseFulfilledResult<GitCommandResult>).value.stdout.trim();
|
||||
const snapshot: GitWorkspaceSnapshot = {
|
||||
headCommit: head, branchName: branch && branch !== "HEAD" ? branch : null,
|
||||
overlayPaths: writer.paths("overlay"), deletedPaths: writer.paths("deleted"), ignoredPaths: writer.paths("ignored"),
|
||||
...(repositories.length ? { repositories } : {}),
|
||||
};
|
||||
writer.close();
|
||||
ownedSnapshots.set(snapshot, path.dirname(writer.filePath));
|
||||
return snapshot;
|
||||
} catch (error) {
|
||||
writer.close(false);
|
||||
for (const repo of repositories) await disposeGitWorkspaceSnapshot(repo.snapshot);
|
||||
await fs.rm(path.dirname(writer.filePath), { recursive: true, force: true });
|
||||
throw error;
|
||||
} finally { options.signal?.removeEventListener("abort", abort); }
|
||||
}
|
||||
|
||||
/** The `git ls-files --others --ignored` output for one directory, read by {@link readReferencedSourceGitIgnoredPaths}. */
|
||||
@@ -541,6 +572,17 @@ export async function readSanitizedOriginRemoteUrl(localDir: string): Promise<st
|
||||
}
|
||||
}
|
||||
|
||||
async function copyCloneTree(source: string, target: string): Promise<void> {
|
||||
await fs.mkdir(target, { recursive: true });
|
||||
for await (const entry of await fs.opendir(source)) {
|
||||
const from = path.join(source, entry.name);
|
||||
const to = path.join(target, entry.name);
|
||||
if (entry.isDirectory()) await copyCloneTree(from, to);
|
||||
else if (entry.isSymbolicLink()) await fs.symlink(await fs.readlink(from), to);
|
||||
else { await fs.copyFile(from, to); await fs.chmod(to, (await fs.stat(from)).mode); }
|
||||
}
|
||||
}
|
||||
|
||||
export async function withShallowGitWorkspaceClone<T>(
|
||||
input: {
|
||||
localDir: string;
|
||||
@@ -599,10 +641,7 @@ export async function withShallowGitWorkspaceClone<T>(
|
||||
// Preserve repository-relative links. fs.cp otherwise rewrites them to
|
||||
// absolute paths into nestedClone, which is deleted after this callback
|
||||
// and is outside the workspace when the sandbox restores its files.
|
||||
await fs.cp(nestedClone, path.join(cloneDir, repository.path), {
|
||||
recursive: true,
|
||||
verbatimSymlinks: true,
|
||||
});
|
||||
await copyCloneTree(nestedClone, path.join(cloneDir, repository.path));
|
||||
});
|
||||
}
|
||||
if (input.snapshot.repositories?.length) {
|
||||
@@ -751,11 +790,13 @@ export function buildRemoteGitDeltaBundleScript(input: {
|
||||
"fi",
|
||||
statusPath
|
||||
? [
|
||||
`if [ -z "$(git -C ${remoteDir} status --porcelain=v1 --untracked-files=normal)" ]; then`,
|
||||
`git -C ${remoteDir} status --porcelain=v1 --untracked-files=normal -z > ${shellQuote(`${input.statusPath}.git-status`)}`,
|
||||
`if [ ! -s ${shellQuote(`${input.statusPath}.git-status`)} ]; then`,
|
||||
` printf clean > ${statusPath}`,
|
||||
"else",
|
||||
` printf dirty > ${statusPath}`,
|
||||
"fi",
|
||||
`rm -f -- ${shellQuote(`${input.statusPath}.git-status`)}`,
|
||||
].join("\n")
|
||||
: "",
|
||||
input.catBundle ? `cat ${bundlePath}` : "",
|
||||
@@ -817,8 +858,13 @@ export async function integrateImportedGitHead(input: {
|
||||
};
|
||||
|
||||
for (let attempt = 0; attempt < 5; attempt += 1) {
|
||||
const snapshot = await readGitWorkspaceSnapshot(input.localDir);
|
||||
if (!snapshot) return;
|
||||
const snapshot = {
|
||||
headCommit: (await runLocalGit(input.localDir, ["rev-parse", "HEAD"])).stdout.trim(),
|
||||
branchName: (await runLocalGit(input.localDir, ["symbolic-ref", "--quiet", "--short", "HEAD"]).catch((error) => {
|
||||
if (error.code === 1) return { stdout: "" };
|
||||
throw error;
|
||||
})).stdout.trim() || null,
|
||||
};
|
||||
|
||||
const currentHead = snapshot.headCommit;
|
||||
if (!currentHead || currentHead === input.importedHead) return;
|
||||
@@ -947,25 +993,14 @@ export async function resetLocalGitIndexToHead(input: {
|
||||
throw new Error(`Failed to reset local git index to HEAD after workspace restore: ${detail}`);
|
||||
}
|
||||
|
||||
const stagedDiff = await runLocalGit(input.localDir, ["diff", "--cached", "--name-status", "HEAD", "--"], {
|
||||
timeout: 10_000,
|
||||
maxBuffer: 1024 * 1024,
|
||||
});
|
||||
if (stagedDiff.stdout.trim().length > 0) {
|
||||
throw new Error(
|
||||
`Workspace restore left staged git index changes after reset:\n${stagedDiff.stdout.trim()}`,
|
||||
);
|
||||
const hasDiff = async (args: string[]) => {
|
||||
try { await runLocalGit(input.localDir, args, { timeout: workspaceSnapshotTimeoutMs(), maxBuffer: 64 * 1024 }); return false; }
|
||||
catch (error) { if ((error as { code?: unknown }).code === 1) return true; throw error; }
|
||||
};
|
||||
if (await hasDiff(["diff", "--quiet", "--cached", "HEAD", "--"])) {
|
||||
throw new Error("Workspace restore left staged git index changes after reset");
|
||||
}
|
||||
|
||||
if (!input.checkWorkingTreeClean) return;
|
||||
|
||||
const workingTreeDiff = await runLocalGit(input.localDir, ["diff", "--name-status", "HEAD", "--"], {
|
||||
timeout: 10_000,
|
||||
maxBuffer: 1024 * 1024,
|
||||
});
|
||||
if (workingTreeDiff.stdout.trim().length > 0) {
|
||||
console.warn(
|
||||
"[paperclip] Workspace restore preserved local working tree changes after clean sandbox restore.",
|
||||
);
|
||||
if (input.checkWorkingTreeClean && await hasDiff(["diff", "--quiet", "HEAD", "--"])) {
|
||||
console.warn("[paperclip] Workspace restore preserved local working tree changes after clean sandbox restore.");
|
||||
}
|
||||
}
|
||||
@@ -1673,7 +1673,7 @@ describe("sandbox managed runtime", () => {
|
||||
expect(downloadedTars).toHaveLength(1);
|
||||
const members = await listTarMembers(rootDir, "empty-workspace-download.tar", downloadedTars[0]!.bytes);
|
||||
expect(members).toEqual([]);
|
||||
const emptyArchiveCommand = runCommands.find((command) => command.includes("dd if=/dev/zero"));
|
||||
const emptyArchiveCommand = runCommands.find((command) => command.includes("--null -T"));
|
||||
expect(emptyArchiveCommand).toBeDefined();
|
||||
expect(emptyArchiveCommand).not.toContain("/dev/null");
|
||||
});
|
||||
|
||||
@@ -6,6 +6,7 @@ import {
|
||||
promises as fs,
|
||||
} from "node:fs";
|
||||
import os from "node:os";
|
||||
import { workspacePaths, workspacePathMatcher, writeWorkspacePaths, isPathManifest, type WorkspacePaths } from "./workspace-manifest.js";
|
||||
import path from "node:path";
|
||||
import { promisify } from "node:util";
|
||||
import {
|
||||
@@ -19,6 +20,7 @@ import {
|
||||
type GitWorkspaceSnapshot,
|
||||
integrateImportedGitHead,
|
||||
readGitWorkspaceSnapshot,
|
||||
disposeGitWorkspaceSnapshot,
|
||||
ReferencedSourceIgnoreScanLimitExceededError,
|
||||
readReferencedSourceGitIgnoredPaths,
|
||||
resetLocalGitIndexToHead,
|
||||
@@ -27,6 +29,8 @@ import {
|
||||
} from "./git-workspace-sync.js";
|
||||
import {
|
||||
captureDirectorySnapshot,
|
||||
disposeDirectorySnapshot,
|
||||
selectDirectorySnapshot,
|
||||
mergeDirectoryWithBaseline,
|
||||
type DirectorySnapshot,
|
||||
} from "./workspace-restore-merge.js";
|
||||
@@ -568,6 +572,7 @@ export interface PreparedSandboxManagedRuntime {
|
||||
baseline: DirectorySnapshot;
|
||||
gitSnapshot: GitWorkspaceSnapshot | null;
|
||||
} | null;
|
||||
cleanupWorkspaceSnapshot(): Promise<void>;
|
||||
restoreWorkspace(onProgress?: RuntimeProgressSink): Promise<void>;
|
||||
}
|
||||
|
||||
@@ -654,16 +659,22 @@ function buildWorkspaceTarExtractCommand(input: {
|
||||
// Named builder (C3): remove paths deleted in the host git worktree from the
|
||||
// sandbox workspace. Every path is shell-quoted; the caller supplies only
|
||||
// already-confined relative paths from the git snapshot.
|
||||
function buildRemoveDeletedPathsCommand(input: {
|
||||
remoteDir: string;
|
||||
deletedPaths: string[];
|
||||
}): string {
|
||||
const quotedPaths = input.deletedPaths.map((entry) => shellQuote(entry)).join(" ");
|
||||
return `cd ${shellQuote(input.remoteDir)} && rm -rf -- ${quotedPaths}`;
|
||||
function buildRemoveDeletedPathsCommand(input: { remoteDir: string; manifestPath: string }): string {
|
||||
// NUL input plus bounded xargs batches preserve whitespace and never create
|
||||
// one argument list for the full snapshot. Refuse symlink ancestors.
|
||||
const remove = `for entry do
|
||||
parent=$entry
|
||||
while [ "\${parent#*/}" != "$parent" ]; do
|
||||
parent=\${parent%/*}
|
||||
if [ -L "$parent" ] || { [ -e "$parent" ] && [ ! -d "$parent" ]; }; then exit 42; fi
|
||||
done
|
||||
rm -rf -- "$entry" || exit
|
||||
done`;
|
||||
return `cd ${shellQuote(input.remoteDir)} && xargs -0 -r -n 64 sh -c ${shellQuote(remove)} sh < ${shellQuote(input.manifestPath)} && rm -f -- ${shellQuote(input.manifestPath)}`;
|
||||
}
|
||||
|
||||
function buildUniqueStagingPath(input: { targetPath: string; suffix: string }): string {
|
||||
return `${input.targetPath}${input.suffix}.${randomUUID()}`;
|
||||
return path.join(path.dirname(input.targetPath), `${input.suffix}.${randomUUID()}`);
|
||||
}
|
||||
|
||||
// The workspace stages under `<runtimeRootDir>/workspace-upload.tar` and, for a
|
||||
@@ -825,32 +836,15 @@ export async function createTarballFromDirectory(input: {
|
||||
// entries avoids the self-entry entirely and is portable across GNU/BSD/busybox
|
||||
// tar (no GNU-only --no-overwrite-dir needed). --exclude still filters nested
|
||||
// matches and any named entry it matches.
|
||||
const entries = (await fs.readdir(input.localDir)).sort((left, right) => left.localeCompare(right));
|
||||
if (entries.length === 0) {
|
||||
// A workspace can legitimately be empty (blank-workspace agent runs). Write a
|
||||
// valid empty tar archive (1024-byte all-zero EOF marker) so extraction is a
|
||||
// clean no-op rather than tar refusing to create an empty archive.
|
||||
await fs.writeFile(input.archivePath, Buffer.alloc(1024));
|
||||
return;
|
||||
}
|
||||
await execTar([
|
||||
"-c",
|
||||
// Prevent macOS bsdtar from embedding LIBARCHIVE.xattr.* PAX extended
|
||||
// headers for extended attributes (e.g. com.apple.provenance). GNU tar on
|
||||
// Linux does not recognise these proprietary headers and fails extraction
|
||||
// with "This does not look like a tar archive". COPYFILE_DISABLE=1 (set in
|
||||
// execTar) already suppresses AppleDouble ._* sidecar files; --no-xattrs
|
||||
// additionally suppresses the inline PAX xattr entries.
|
||||
"--no-xattrs",
|
||||
...(input.followSymlinks ? ["-h"] : []),
|
||||
"-f",
|
||||
input.archivePath,
|
||||
"-C",
|
||||
input.localDir,
|
||||
...excludeArgs,
|
||||
"--",
|
||||
...entries,
|
||||
]);
|
||||
await withTempDir("paperclip-tar-list-", async (directory) => {
|
||||
const list = path.join(directory, "entries.nul");
|
||||
const file = await fs.open(list, "wx", 0o600);
|
||||
try {
|
||||
for await (const entry of await fs.opendir(input.localDir)) await file.writeFile(`${entry.name}\0`);
|
||||
} finally { await file.close(); }
|
||||
await execTar(["-c", "--no-xattrs", ...(input.followSymlinks ? ["-h"] : []),
|
||||
"-f", input.archivePath, "-C", input.localDir, ...excludeArgs, "--null", "-T", list]);
|
||||
});
|
||||
}
|
||||
|
||||
async function extractTarballToDirectory(input: {
|
||||
@@ -861,18 +855,13 @@ async function extractTarballToDirectory(input: {
|
||||
await execTar(["-xf", input.archivePath, "-C", input.localDir]);
|
||||
}
|
||||
|
||||
async function walkDirectory(root: string, relative = ""): Promise<string[]> {
|
||||
const current = path.join(root, relative);
|
||||
const entries = await fs.readdir(current, { withFileTypes: true }).catch(() => []);
|
||||
const out: string[] = [];
|
||||
for (const entry of entries) {
|
||||
async function* walkDirectory(root: string, relative = "", childrenFirst = false): AsyncGenerator<string> {
|
||||
for await (const entry of await fs.opendir(path.join(root, relative))) {
|
||||
const nextRelative = relative ? path.posix.join(relative, entry.name) : entry.name;
|
||||
out.push(nextRelative);
|
||||
if (entry.isDirectory()) {
|
||||
out.push(...(await walkDirectory(root, nextRelative)));
|
||||
}
|
||||
if (!childrenFirst) yield nextRelative;
|
||||
if (entry.isDirectory()) yield* walkDirectory(root, nextRelative, childrenFirst);
|
||||
if (childrenFirst) yield nextRelative;
|
||||
}
|
||||
return out.sort((left, right) => right.length - left.length);
|
||||
}
|
||||
|
||||
async function copyWorkspaceEntry(sourceRoot: string, targetRoot: string, relative: string): Promise<void> {
|
||||
@@ -916,17 +905,18 @@ export async function mirrorDirectory(
|
||||
const shouldPreserveAbsent = (relative: string) =>
|
||||
[...preserveAbsent].some((candidate) => isRelativePathOrDescendant(relative, candidate));
|
||||
|
||||
const sourceEntries = new Set(await walkDirectory(sourceDir));
|
||||
const targetEntries = await walkDirectory(targetDir);
|
||||
for (const relative of targetEntries) {
|
||||
for await (const relative of walkDirectory(targetDir, "", true)) {
|
||||
if (shouldPreserveAbsent(relative)) continue;
|
||||
if (!sourceEntries.has(relative)) {
|
||||
const sourceExists = await fs.lstat(path.join(sourceDir, relative)).then(() => true).catch((error: NodeJS.ErrnoException) => {
|
||||
if (error.code === "ENOENT" || error.code === "ENOTDIR") return false;
|
||||
throw error;
|
||||
});
|
||||
if (!sourceExists) {
|
||||
await fs.rm(path.join(targetDir, relative), { recursive: true, force: true }).catch(() => undefined);
|
||||
}
|
||||
}
|
||||
|
||||
const entries = (await walkDirectory(sourceDir)).sort((left, right) => left.localeCompare(right));
|
||||
for (const relative of entries) {
|
||||
for await (const relative of walkDirectory(sourceDir)) {
|
||||
await copyWorkspaceEntry(sourceDir, targetDir, relative);
|
||||
}
|
||||
}
|
||||
@@ -947,7 +937,8 @@ async function captureWorkspaceSourceRoot(localDir: string): Promise<WorkspaceSo
|
||||
async function copySelectedWorkspaceEntries(input: {
|
||||
sourceRoot: WorkspaceSourceRoot;
|
||||
targetDir: string;
|
||||
relativePaths: string[];
|
||||
relativePaths: WorkspacePaths;
|
||||
ignoredPaths?: WorkspacePaths;
|
||||
exclude: string[];
|
||||
}): Promise<void> {
|
||||
await fs.mkdir(input.targetDir, { recursive: true });
|
||||
@@ -959,8 +950,10 @@ async function copySelectedWorkspaceEntries(input: {
|
||||
}
|
||||
};
|
||||
await assertSourceRoot();
|
||||
for (const relative of input.relativePaths) {
|
||||
if (shouldExcludePath(relative, input.exclude)) continue;
|
||||
const ignored = workspacePathMatcher(input.ignoredPaths);
|
||||
try {
|
||||
for (const relative of workspacePaths(input.relativePaths)) {
|
||||
if (shouldExcludePath(relative, input.exclude) || ignored.matches(relative)) continue;
|
||||
const sourcePath = path.join(sourceDir, relative);
|
||||
const parentSegments = path.relative(sourceDir, path.dirname(sourcePath)).split(path.sep).filter(Boolean);
|
||||
const assertParentDirectory = async () => {
|
||||
@@ -992,6 +985,7 @@ async function copySelectedWorkspaceEntries(input: {
|
||||
await assertSourceRoot();
|
||||
await assertParentDirectory();
|
||||
}
|
||||
} finally { ignored.close(); }
|
||||
}
|
||||
|
||||
function toBuffer(bytes: Buffer | Uint8Array | ArrayBuffer): Buffer {
|
||||
@@ -1019,17 +1013,13 @@ function createRemoteTarballFromDirectoryCommand(input: {
|
||||
archivePath: string;
|
||||
exclude?: string[];
|
||||
}): string {
|
||||
// Match the local archive path: name top-level entries explicitly so tar
|
||||
// does not include a "." self-entry that it later tries to chmod/utime.
|
||||
const list = `${input.archivePath}.entries`;
|
||||
return [
|
||||
`mkdir -p ${shellQuote(path.posix.dirname(input.archivePath))}`,
|
||||
`cd ${shellQuote(input.remoteDir)}`,
|
||||
"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(input.archivePath)} bs=1024 count=1; ` +
|
||||
`else tar -cf ${shellQuote(input.archivePath)} ${tarExcludeFlags(input.exclude)} -- "$@"; fi`,
|
||||
`find . -mindepth 1 -maxdepth 1 -print0 > ${shellQuote(list)}`,
|
||||
`tar -cf ${shellQuote(input.archivePath)} ${tarExcludeFlags(input.exclude)} --null -T ${shellQuote(list)}`,
|
||||
`rm -f -- ${shellQuote(list)}`,
|
||||
].join(" && ");
|
||||
}
|
||||
|
||||
@@ -1196,6 +1186,12 @@ export async function prepareSandboxManagedRuntime(input: {
|
||||
readGitWorkspaceSnapshot(input.workspaceLocalDir),
|
||||
)
|
||||
: null;
|
||||
let baselineSnapshot: DirectorySnapshot | null = null;
|
||||
const cleanupWorkspaceSnapshot = async () => {
|
||||
await disposeDirectorySnapshot(baselineSnapshot);
|
||||
await disposeGitWorkspaceSnapshot(gitSnapshot);
|
||||
};
|
||||
try {
|
||||
// A selected subfolder has no cloneable Git snapshot, but its parent
|
||||
// repository's ignore rules still govern which files may leave the host.
|
||||
// Use the same bounded, path-relative resolver as referenced project trees.
|
||||
@@ -1205,8 +1201,7 @@ export async function prepareSandboxManagedRuntime(input: {
|
||||
if (directoryIgnore?.kind === "failed") {
|
||||
throw new Error(`Workspace ignore scan failed: ${directoryIgnore.reason}`);
|
||||
}
|
||||
const gitIgnoredExcludes = gitSnapshot?.ignoredPaths
|
||||
?? (directoryIgnore?.kind === "git" ? directoryIgnore.ignoredPaths : undefined);
|
||||
const gitIgnoredExcludes = directoryIgnore?.kind === "git" ? directoryIgnore.ignoredPaths : undefined;
|
||||
const workspaceArchiveExclude = mergeExcludes(
|
||||
SANDBOX_WORKSPACE_HEAVY_DIR_EXCLUDES,
|
||||
[...GIT_ARCHIVE_EXCLUDES],
|
||||
@@ -1227,11 +1222,13 @@ export async function prepareSandboxManagedRuntime(input: {
|
||||
// `lstat`s every entry and SHA-256-hashes every file's bytes. This is the
|
||||
// dominant cost in the pre-`pack` window — it reads the content of every
|
||||
// non-excluded file, serially — so it earns its own span.
|
||||
const baselineSnapshot = syncWorkspace
|
||||
baselineSnapshot = syncWorkspace
|
||||
? (input.workspaceBaseline ??
|
||||
(await runStepSpan("snapshot.baseline", () =>
|
||||
captureDirectorySnapshot(input.workspaceLocalDir, {
|
||||
exclude: restoreExclude,
|
||||
ignoredPaths: gitSnapshot?.ignoredPaths,
|
||||
diskBacked: true,
|
||||
}),
|
||||
)))
|
||||
: null;
|
||||
@@ -1461,6 +1458,7 @@ export async function prepareSandboxManagedRuntime(input: {
|
||||
sourceRoot: workspaceRoot!,
|
||||
targetDir: workspaceArchiveDir,
|
||||
relativePaths: gitSnapshot.overlayPaths,
|
||||
ignoredPaths: gitSnapshot.ignoredPaths,
|
||||
exclude: workspaceArchiveExclude,
|
||||
});
|
||||
}
|
||||
@@ -1495,13 +1493,12 @@ export async function prepareSandboxManagedRuntime(input: {
|
||||
}),
|
||||
});
|
||||
// 3. Optional remove-deleted-paths command runs LAST, after both extracts.
|
||||
if (gitSnapshot && gitSnapshot.deletedPaths.length > 0) {
|
||||
workspacePostUploadCommands.push({
|
||||
command: buildRemoveDeletedPathsCommand({
|
||||
remoteDir: workspaceRemoteDir,
|
||||
deletedPaths: gitSnapshot.deletedPaths,
|
||||
}),
|
||||
});
|
||||
if (gitSnapshot && (isPathManifest(gitSnapshot.deletedPaths) ? gitSnapshot.deletedPaths.count : gitSnapshot.deletedPaths.length) > 0) {
|
||||
const deletedManifest = path.join(tempDir, "deleted.nul");
|
||||
const remoteDeletedManifest = path.posix.join(runtimeRootDir, "deleted.nul");
|
||||
await writeWorkspacePaths(gitSnapshot.deletedPaths, deletedManifest);
|
||||
workspaceFiles.push({ sourcePath: deletedManifest, targetPath: remoteDeletedManifest, kind: "file", access: "rw", writablePath: runtimeRootDir });
|
||||
workspacePostUploadCommands.push({ command: buildRemoveDeletedPathsCommand({ remoteDir: workspaceRemoteDir, manifestPath: remoteDeletedManifest }) });
|
||||
}
|
||||
workspaceUploadBytes += (await fs.stat(workspaceTarPath)).size;
|
||||
});
|
||||
@@ -1704,7 +1701,9 @@ export async function prepareSandboxManagedRuntime(input: {
|
||||
syncWorkspace && baselineSnapshot
|
||||
? { baseline: baselineSnapshot, gitSnapshot }
|
||||
: null,
|
||||
cleanupWorkspaceSnapshot,
|
||||
restoreWorkspace: async (onProgress?: RuntimeProgressSink) => {
|
||||
try {
|
||||
const restoreSink = onProgress ?? input.onProgress;
|
||||
|
||||
// Build the ordered outbound restore task list. Each task runs one
|
||||
@@ -1739,7 +1738,6 @@ export async function prepareSandboxManagedRuntime(input: {
|
||||
throw new Error("Project repository escaped its workspace");
|
||||
}
|
||||
const nestedExclude = mergeExcludes(
|
||||
repository.snapshot.ignoredPaths,
|
||||
baselineSnapshot!.exclude.flatMap((entry) =>
|
||||
entry.startsWith(prefix) ? [entry.slice(prefix.length)]
|
||||
: entry.startsWith("*/") ? [entry] : []),
|
||||
@@ -1752,17 +1750,11 @@ export async function prepareSandboxManagedRuntime(input: {
|
||||
workspaceInboundMode: "adopt_remote",
|
||||
workspaceGitSnapshot: repository.snapshot,
|
||||
workspaceExclude: nestedExclude,
|
||||
workspaceBaseline: {
|
||||
exclude: mergeExcludes(
|
||||
SANDBOX_WORKSPACE_HEAVY_DIR_EXCLUDES,
|
||||
[...GIT_ARCHIVE_EXCLUDES],
|
||||
[".paperclip-runtime"],
|
||||
nestedExclude,
|
||||
),
|
||||
entries: new Map([...baselineSnapshot!.entries]
|
||||
.filter(([entry]) => entry.startsWith(prefix))
|
||||
.map(([entry, value]) => [entry.slice(prefix.length), value])),
|
||||
},
|
||||
workspaceBaseline: await selectDirectorySnapshot(baselineSnapshot!, {
|
||||
prefix,
|
||||
exclude: mergeExcludes(SANDBOX_WORKSPACE_HEAVY_DIR_EXCLUDES, [...GIT_ARCHIVE_EXCLUDES], [".paperclip-runtime"], nestedExclude),
|
||||
ignoredPaths: repository.snapshot.ignoredPaths,
|
||||
}),
|
||||
onRuntimeProgress: input.onRuntimeProgress,
|
||||
});
|
||||
await nested.restoreWorkspace(restoreSink);
|
||||
@@ -1923,12 +1915,12 @@ export async function prepareSandboxManagedRuntime(input: {
|
||||
});
|
||||
}
|
||||
const gitHeadToIntegrate = importedHead;
|
||||
const mergeBaseline = repositories.length === 0 ? baselineSnapshot! : await selectDirectorySnapshot(baselineSnapshot!, {
|
||||
omit: repositories.map((repo) => repo.path), exclude: workspaceRestoreExclude, ignoredPaths: gitSnapshot?.ignoredPaths,
|
||||
});
|
||||
try {
|
||||
await mergeDirectoryWithBaseline({
|
||||
baseline: repositories.length === 0 ? baselineSnapshot! : {
|
||||
exclude: workspaceRestoreExclude,
|
||||
entries: new Map([...baselineSnapshot!.entries].filter(([entry]) =>
|
||||
!repositories.some((repo) => entry === repo.path || entry.startsWith(`${repo.path}/`)))),
|
||||
},
|
||||
baseline: mergeBaseline,
|
||||
sourceDir: extractedDir,
|
||||
targetDir: input.workspaceLocalDir,
|
||||
beforeApply: gitHeadToIntegrate
|
||||
@@ -1948,6 +1940,7 @@ export async function prepareSandboxManagedRuntime(input: {
|
||||
}
|
||||
: undefined,
|
||||
});
|
||||
} finally { if (mergeBaseline !== baselineSnapshot) await disposeDirectorySnapshot(mergeBaseline); }
|
||||
} finally {
|
||||
await emitRuntimeStatus(input.onRuntimeProgress, "finalize", "Finalizing workspace");
|
||||
if (importedRef) {
|
||||
@@ -1999,6 +1992,8 @@ export async function prepareSandboxManagedRuntime(input: {
|
||||
throw result.reason;
|
||||
}
|
||||
}
|
||||
} finally { await cleanupWorkspaceSnapshot(); }
|
||||
},
|
||||
};
|
||||
} catch (error) { await cleanupWorkspaceSnapshot(); throw error; }
|
||||
}
|
||||
@@ -0,0 +1,93 @@
|
||||
import { spawn, type ChildProcess } from "node:child_process";
|
||||
import { WORKSPACE_STREAM_CHUNK_BYTES } from "./workspace-manifest.js";
|
||||
|
||||
export interface WorkspaceGitProcessInput {
|
||||
cwd: string;
|
||||
args: readonly string[];
|
||||
env?: NodeJS.ProcessEnv;
|
||||
signal?: AbortSignal;
|
||||
timeoutMs: number;
|
||||
killGraceMs?: number;
|
||||
maxStdoutBytes: number;
|
||||
maxStderrBytes: number;
|
||||
onStdout?: (chunk: Buffer) => Promise<void> | void;
|
||||
gitBinary?: string;
|
||||
gitArgsPrefix?: readonly string[];
|
||||
}
|
||||
|
||||
function failure(code: string, message: string, details: Record<string, unknown> = {}): Error {
|
||||
return Object.assign(new Error(message), { code, details });
|
||||
}
|
||||
|
||||
function signalProcess(child: ChildProcess, signal: NodeJS.Signals): void {
|
||||
// The leader may have exited while a descendant still holds its pipes open.
|
||||
if (process.platform !== "win32" && child.pid) {
|
||||
try { process.kill(-child.pid, signal); return; } catch { /* direct child fallback */ }
|
||||
}
|
||||
try { child.kill(signal); } catch { /* close owns settlement */ }
|
||||
}
|
||||
|
||||
/** Completion is a barrier for the process, its pipes, and the awaited sink. */
|
||||
export async function runWorkspaceGitProcess(input: WorkspaceGitProcessInput): Promise<{ stdout: string; stderr: string }> {
|
||||
if (input.signal?.aborted) throw failure("workspace_git_scan_cancelled", "Workspace Git scan was cancelled");
|
||||
const child = spawn(input.gitBinary ?? "git", [...(input.gitArgsPrefix ?? []), "-C", input.cwd, ...input.args], {
|
||||
cwd: input.cwd, env: input.env ?? process.env,
|
||||
stdio: ["ignore", "pipe", "pipe"], detached: process.platform !== "win32", windowsHide: true,
|
||||
});
|
||||
let error: Error | null = null;
|
||||
let killTimer: NodeJS.Timeout | undefined;
|
||||
const terminate = (reason: Error) => {
|
||||
if (error) return;
|
||||
error = reason;
|
||||
signalProcess(child, "SIGTERM");
|
||||
killTimer = setTimeout(() => signalProcess(child, "SIGKILL"), input.killGraceMs ?? 250);
|
||||
killTimer.unref();
|
||||
};
|
||||
const onAbort = () => terminate(failure("workspace_git_scan_cancelled", "Workspace Git scan was cancelled"));
|
||||
input.signal?.addEventListener("abort", onAbort, { once: true });
|
||||
if (input.signal?.aborted) onAbort();
|
||||
const timeout = setTimeout(() => terminate(failure("workspace_git_scan_timeout", `Workspace Git scan timed out after ${input.timeoutMs}ms`, { timeoutMs: input.timeoutMs })), input.timeoutMs);
|
||||
timeout.unref();
|
||||
const closed = new Promise<{ code: number | null; signal: NodeJS.Signals | null }>((resolve) => {
|
||||
child.once("error", (cause) => terminate(failure("workspace_git_scan_failed", "Workspace Git scan could not start", { cause: cause.message })));
|
||||
child.once("close", (code, signal) => resolve({ code, signal }));
|
||||
});
|
||||
const collect = async (stream: NonNullable<typeof child.stdout>, limit: number, sink?: WorkspaceGitProcessInput["onStdout"]) => {
|
||||
const chunks: Buffer[] = [];
|
||||
let bytes = 0;
|
||||
try {
|
||||
for await (const raw of stream) {
|
||||
const buffer = Buffer.isBuffer(raw) ? raw : Buffer.from(raw);
|
||||
if (error) continue;
|
||||
if (sink) {
|
||||
for (let offset = 0; offset < buffer.length; offset += WORKSPACE_STREAM_CHUNK_BYTES) {
|
||||
if (error) break;
|
||||
await sink(buffer.subarray(offset, offset + WORKSPACE_STREAM_CHUNK_BYTES));
|
||||
}
|
||||
} else {
|
||||
bytes += buffer.length;
|
||||
if (bytes > limit) {
|
||||
terminate(failure("workspace_git_scan_output_limit", "Workspace Git scan exceeded its output limit"));
|
||||
} else chunks.push(buffer);
|
||||
}
|
||||
}
|
||||
} catch (cause) {
|
||||
terminate(failure("workspace_git_scan_failed", "Workspace Git scan stream failed", { cause: cause instanceof Error ? cause.message : String(cause) }));
|
||||
}
|
||||
return Buffer.concat(chunks).toString("utf8");
|
||||
};
|
||||
try {
|
||||
const [outcome, stdout, stderr] = await Promise.all([
|
||||
closed, collect(child.stdout!, input.maxStdoutBytes, input.onStdout), collect(child.stderr!, input.maxStderrBytes),
|
||||
]);
|
||||
if (error) throw error;
|
||||
if (outcome.code !== 0) throw failure("workspace_git_scan_failed", "Workspace Git scan failed", {
|
||||
exitCode: outcome.code, signal: outcome.signal, stderr: stderr.trim().slice(0, 1000),
|
||||
});
|
||||
return { stdout, stderr };
|
||||
} finally {
|
||||
clearTimeout(timeout);
|
||||
if (killTimer) clearTimeout(killTimer);
|
||||
input.signal?.removeEventListener("abort", onAbort);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,95 @@
|
||||
import fs from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterEach, expect, it, vi } from "vitest";
|
||||
import { createWorkspaceManifest, readManifestRecords, WorkspaceNulParser, WORKSPACE_PATH_MAX_BYTES, workspacePaths } from "./workspace-manifest.js";
|
||||
import { readGitWorkspaceSnapshot, disposeGitWorkspaceSnapshot, runLocalGit, setExpensiveWorkspaceGitExecutor } from "./git-workspace-sync.js";
|
||||
|
||||
const cleanup: string[] = [];
|
||||
afterEach(async () => {
|
||||
setExpensiveWorkspaceGitExecutor(null);
|
||||
vi.restoreAllMocks();
|
||||
for (const dir of cleanup.splice(0)) await fs.rm(dir, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
it("parses split UTF-8 and NUL records while rejecting truncation and oversized records", () => {
|
||||
const names: string[] = [];
|
||||
const parser = new WorkspaceNulParser((value) => names.push(value));
|
||||
const bytes = Buffer.from(" 雪\n\t \0-option\0");
|
||||
for (const byte of bytes) parser.write(Buffer.from([byte]));
|
||||
parser.finish();
|
||||
expect(names).toEqual([" 雪\n\t ", "-option"]);
|
||||
parser.write(Buffer.from("unfinished"));
|
||||
expect(() => parser.finish()).toThrow("Incomplete");
|
||||
expect(() => parser.write(Buffer.alloc(WORKSPACE_PATH_MAX_BYTES, 97))).toThrow("per-record");
|
||||
expect(() => new WorkspaceNulParser(() => {}).write(Buffer.from([255, 0]))).toThrow();
|
||||
});
|
||||
|
||||
it("deduplicates paths on disk and rejects incomplete manifest metadata", async () => {
|
||||
const writer = await createWorkspaceManifest();
|
||||
cleanup.push(path.dirname(writer.filePath));
|
||||
writer.batch(() => { writer.add("overlay", "b"); writer.add("overlay", "a"); writer.add("overlay", "b"); });
|
||||
const manifest = writer.paths("overlay");
|
||||
writer.close();
|
||||
expect([...workspacePaths(manifest)]).toEqual(["a", "b"]);
|
||||
expect(() => [...readManifestRecords({ ...manifest, count: 3 })]).toThrow("Incomplete");
|
||||
expect(() => [...workspacePaths(["../escape"])]).toThrow("Invalid");
|
||||
});
|
||||
|
||||
it("settles a late producer before removing shared storage after another scan fails", async () => {
|
||||
const repo = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-scan-barrier-"));
|
||||
cleanup.push(repo);
|
||||
await runLocalGit(repo, ["init"]);
|
||||
await runLocalGit(repo, ["-c", "user.name=Test", "-c", "user.email=test@example.test", "commit", "--allow-empty", "-qm", "fixture"], {
|
||||
env: { ...process.env, GIT_AUTHOR_NAME: "Test", GIT_COMMITTER_NAME: "Test", GIT_AUTHOR_EMAIL: "test@example.test", GIT_COMMITTER_EMAIL: "test@example.test" },
|
||||
});
|
||||
let release!: () => void;
|
||||
const gate = new Promise<void>((resolve) => { release = resolve; });
|
||||
let lateFinished = false;
|
||||
let settled = false;
|
||||
const removed: string[] = [];
|
||||
const originalRm = fs.rm.bind(fs);
|
||||
vi.spyOn(fs, "rm").mockImplementation(async (target, options) => {
|
||||
if (String(target).includes("paperclip-workspace-manifest-")) {
|
||||
expect(lateFinished).toBe(true);
|
||||
removed.push(String(target));
|
||||
}
|
||||
return originalRm(target, options);
|
||||
});
|
||||
const failure = Object.assign(new Error("scan failed"), { code: "workspace_git_scan_failed" });
|
||||
setExpensiveWorkspaceGitExecutor(async (input) => {
|
||||
if (input.operation.endsWith("ignored_files")) throw failure;
|
||||
if (input.operation.endsWith("untracked_files")) {
|
||||
await gate;
|
||||
await input.onStdout!(Buffer.from("late-producer\0"));
|
||||
lateFinished = true;
|
||||
}
|
||||
return { stdout: "", stderr: "" };
|
||||
});
|
||||
const result = readGitWorkspaceSnapshot(repo).catch((error) => { settled = true; return error; });
|
||||
await new Promise((resolve) => setTimeout(resolve, 30));
|
||||
expect(settled).toBe(false);
|
||||
expect(removed).toEqual([]);
|
||||
release();
|
||||
expect(await result).toBe(failure);
|
||||
expect(removed).toHaveLength(1);
|
||||
await expect(fs.stat(removed[0]!)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
});
|
||||
|
||||
it("fails a malformed completed scan instead of publishing a partial snapshot", async () => {
|
||||
const repo = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-truncated-scan-"));
|
||||
cleanup.push(repo);
|
||||
await runLocalGit(repo, ["init"]);
|
||||
await runLocalGit(repo, ["-c", "user.name=Test", "-c", "user.email=test@example.test", "commit", "--allow-empty", "-qm", "fixture"], {
|
||||
env: { ...process.env, GIT_AUTHOR_NAME: "Test", GIT_COMMITTER_NAME: "Test", GIT_AUTHOR_EMAIL: "test@example.test", GIT_COMMITTER_EMAIL: "test@example.test" },
|
||||
});
|
||||
setExpensiveWorkspaceGitExecutor(async (input) => {
|
||||
if (input.operation.endsWith("untracked_files")) await input.onStdout!(Buffer.from("complete\0partial"));
|
||||
return { stdout: "", stderr: "" };
|
||||
});
|
||||
await expect(readGitWorkspaceSnapshot(repo)).rejects.toThrow("Incomplete NUL");
|
||||
setExpensiveWorkspaceGitExecutor(null);
|
||||
const recovered = await readGitWorkspaceSnapshot(repo);
|
||||
expect(recovered).not.toBeNull();
|
||||
await disposeGitWorkspaceSnapshot(recovered);
|
||||
});
|
||||
@@ -0,0 +1,184 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import { constants, promises as fs } from "node:fs";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { DatabaseSync, type StatementSync } from "node:sqlite";
|
||||
|
||||
// Limits apply to one record and to SQLite's page cache, never to the list.
|
||||
export const WORKSPACE_PATH_MAX_BYTES = 64 * 1024;
|
||||
export const WORKSPACE_STREAM_CHUNK_BYTES = 64 * 1024;
|
||||
export interface PathManifest {
|
||||
kind: "path_manifest";
|
||||
version: 1;
|
||||
filePath: string;
|
||||
category: string;
|
||||
count: number;
|
||||
}
|
||||
export type WorkspacePaths = readonly string[] | PathManifest;
|
||||
|
||||
export function assertWorkspaceRelativePath(value: string): void {
|
||||
if (!value || value.includes("\0") || path.posix.isAbsolute(value) || path.win32.isAbsolute(value)
|
||||
|| value.split(/[\\/]/).some((part) => part === ".." || part === ".")
|
||||
|| Buffer.byteLength(value) > WORKSPACE_PATH_MAX_BYTES) {
|
||||
throw new Error("Invalid workspace manifest path");
|
||||
}
|
||||
}
|
||||
|
||||
export function isPathManifest(value: WorkspacePaths): value is PathManifest {
|
||||
return !Array.isArray(value);
|
||||
}
|
||||
|
||||
function openDatabase(filePath: string, readOnly = true): DatabaseSync {
|
||||
const db = new DatabaseSync(filePath, { readOnly, allowExtension: false });
|
||||
try {
|
||||
db.exec("PRAGMA trusted_schema=OFF; PRAGMA cache_size=-1024; PRAGMA temp_store=FILE; PRAGMA mmap_size=0;");
|
||||
return db;
|
||||
} catch (error) { db.close(); throw error; }
|
||||
}
|
||||
|
||||
/** One private database, sealed before it is published. Consumers open read-only. */
|
||||
export class WorkspaceManifestWriter {
|
||||
private readonly db: DatabaseSync;
|
||||
private readonly insert: StatementSync;
|
||||
private closed = false;
|
||||
constructor(readonly filePath: string) {
|
||||
this.db = openDatabase(filePath, false);
|
||||
try {
|
||||
this.db.exec("PRAGMA journal_mode=DELETE; CREATE TABLE records(category TEXT, path TEXT, value TEXT NOT NULL, PRIMARY KEY(category,path)) WITHOUT ROWID;");
|
||||
this.insert = this.db.prepare("INSERT OR REPLACE INTO records VALUES(?,?,?)");
|
||||
this.db.exec("BEGIN");
|
||||
} catch (error) { this.db.close(); throw error; }
|
||||
}
|
||||
add(category: string, relative: string, value = ""): void {
|
||||
assertWorkspaceRelativePath(relative);
|
||||
if (Buffer.byteLength(value) > WORKSPACE_PATH_MAX_BYTES) throw new Error("Workspace manifest value exceeds record limit");
|
||||
this.insert.run(category, relative, value);
|
||||
}
|
||||
batch(work: () => void): void {
|
||||
this.db.exec("SAVEPOINT chunk");
|
||||
try { work(); this.db.exec("RELEASE chunk"); }
|
||||
catch (error) { this.db.exec("ROLLBACK TO chunk; RELEASE chunk"); throw error; }
|
||||
}
|
||||
paths(category: string): PathManifest {
|
||||
const row = this.db.prepare("SELECT count(*) AS count FROM records WHERE category=?").get(category)!;
|
||||
return { kind: "path_manifest", version: 1, filePath: this.filePath, category, count: Number(row.count) };
|
||||
}
|
||||
close(commit = true): void {
|
||||
if (this.closed) return;
|
||||
this.closed = true;
|
||||
try { this.db.exec(commit ? "COMMIT" : "ROLLBACK"); }
|
||||
finally { this.db.close(); }
|
||||
}
|
||||
}
|
||||
|
||||
export class WorkspaceManifestMap<T> {
|
||||
private db: DatabaseSync | null = null;
|
||||
private lookup: StatementSync | null = null;
|
||||
constructor(readonly manifest: PathManifest, private readonly parse: (value: string) => T) {}
|
||||
get size(): number { return this.manifest.count; }
|
||||
get(relative: string): T | undefined {
|
||||
this.db ??= openDatabase(this.manifest.filePath);
|
||||
this.lookup ??= this.db.prepare("SELECT value FROM records WHERE category=? AND path=?");
|
||||
const row = this.lookup.get(this.manifest.category, relative);
|
||||
return row ? this.parse(String(row.value)) : undefined;
|
||||
}
|
||||
has(relative: string): boolean { return this.get(relative) !== undefined; }
|
||||
*entries(reverse = false): Generator<[string, T]> {
|
||||
for (const [relative, value] of readManifestRecords(this.manifest, reverse)) yield [relative, this.parse(value)];
|
||||
}
|
||||
[Symbol.iterator](): Generator<[string, T]> { return this.entries(); }
|
||||
close(): void { this.db?.close(); this.db = null; this.lookup = null; }
|
||||
}
|
||||
|
||||
export async function createWorkspaceManifest(prefix = "paperclip-workspace-manifest-"): Promise<WorkspaceManifestWriter> {
|
||||
const directory = await fs.mkdtemp(path.join(os.tmpdir(), prefix));
|
||||
try { return new WorkspaceManifestWriter(path.join(directory, "paths.sqlite")); }
|
||||
catch (error) { await fs.rm(directory, { recursive: true, force: true }); throw error; }
|
||||
}
|
||||
|
||||
export function* readManifestRecords(manifest: PathManifest, reverse = false): Generator<[string, string]> {
|
||||
const db = openDatabase(manifest.filePath);
|
||||
let count = 0;
|
||||
try {
|
||||
for (const row of db.prepare(`SELECT path,value FROM records WHERE category=? ORDER BY path ${reverse ? "DESC" : "ASC"}`).iterate(manifest.category)) {
|
||||
const relative = String(row.path);
|
||||
const value = String(row.value);
|
||||
assertWorkspaceRelativePath(relative);
|
||||
if (Buffer.byteLength(value) > WORKSPACE_PATH_MAX_BYTES) throw new Error("Invalid workspace manifest record");
|
||||
count++;
|
||||
yield [relative, value];
|
||||
}
|
||||
if (count !== manifest.count) throw new Error("Incomplete workspace manifest");
|
||||
} finally { db.close(); }
|
||||
}
|
||||
|
||||
export function* workspacePaths(paths: WorkspacePaths): Generator<string> {
|
||||
if (isPathManifest(paths)) {
|
||||
for (const [relative] of readManifestRecords(paths)) yield relative;
|
||||
} else {
|
||||
for (const relative of paths) { assertWorkspaceRelativePath(relative); yield relative; }
|
||||
}
|
||||
}
|
||||
|
||||
/** Exact paths and their descendants; Git ignore output is literal, not a glob. */
|
||||
export function workspacePathMatcher(paths?: WorkspacePaths): { matches(relative: string): boolean; close(): void } {
|
||||
const db = paths && isPathManifest(paths) ? openDatabase(paths.filePath) : null;
|
||||
const lookup = db?.prepare("SELECT 1 FROM records WHERE category=? AND path=?");
|
||||
return {
|
||||
matches(relative) {
|
||||
let current = relative;
|
||||
while (current) {
|
||||
if (lookup && paths && isPathManifest(paths)
|
||||
? Boolean(lookup.get(paths.category, current))
|
||||
: (paths as readonly string[] | undefined)?.includes(current)) return true;
|
||||
const slash = current.lastIndexOf("/");
|
||||
if (slash < 0) break;
|
||||
current = current.slice(0, slash);
|
||||
}
|
||||
return false;
|
||||
},
|
||||
close() { db?.close(); },
|
||||
};
|
||||
}
|
||||
|
||||
/** Incremental NUL parser: one bounded record, strict UTF-8, no partial success. */
|
||||
export class WorkspaceNulParser {
|
||||
private readonly pending = Buffer.alloc(WORKSPACE_PATH_MAX_BYTES);
|
||||
private length = 0;
|
||||
private readonly decoder = new TextDecoder("utf-8", { fatal: true });
|
||||
constructor(private readonly record: (relative: string) => void) {}
|
||||
write(chunk: Buffer): void {
|
||||
let start = 0;
|
||||
while (start < chunk.length) {
|
||||
const end = chunk.indexOf(0, start);
|
||||
const stop = end < 0 ? chunk.length : end;
|
||||
const size = stop - start;
|
||||
if (this.length + size > this.pending.length) throw new Error("Workspace filename exceeds per-record limit");
|
||||
chunk.copy(this.pending, this.length, start, stop);
|
||||
this.length += size;
|
||||
if (end < 0) return;
|
||||
if (!this.length) throw new Error("Empty workspace filename record");
|
||||
this.record(this.decoder.decode(this.pending.subarray(0, this.length)));
|
||||
this.length = 0;
|
||||
start = end + 1;
|
||||
}
|
||||
}
|
||||
finish(): void { if (this.length) throw new Error("Incomplete NUL-delimited workspace scan"); }
|
||||
}
|
||||
|
||||
export async function writeWorkspacePaths(paths: WorkspacePaths, target: string): Promise<void> {
|
||||
const file = await fs.open(target, "wx", 0o600);
|
||||
try {
|
||||
for (const relative of workspacePaths(paths)) await file.writeFile(`${relative}\0`);
|
||||
} finally { await file.close(); }
|
||||
}
|
||||
|
||||
export async function manifestFileSha256(filePath: string): Promise<string> {
|
||||
const file = await fs.open(filePath, constants.O_RDONLY | constants.O_NOFOLLOW);
|
||||
try {
|
||||
if (!(await file.stat()).isFile()) throw new Error("Invalid workspace manifest file");
|
||||
const digest = createHash("sha256");
|
||||
for await (const chunk of file.createReadStream({ autoClose: false })) digest.update(chunk);
|
||||
return digest.digest("hex");
|
||||
} finally { await file.close(); }
|
||||
}
|
||||
@@ -41,6 +41,7 @@ describe("workspace restore merge", () => {
|
||||
|
||||
const snapshot = await captureDirectorySnapshot(rootDir, { exclude: [] });
|
||||
const serialized = serializeDirectorySnapshot(snapshot);
|
||||
if (serialized.version !== 1) throw new Error("Expected legacy in-memory snapshot");
|
||||
const restored = parseDirectorySnapshot(serialized);
|
||||
|
||||
expect(serialized.entries.map(([relativePath]) => relativePath)).toEqual([
|
||||
|
||||
@@ -2,6 +2,7 @@ import { createHash } from "node:crypto";
|
||||
import { createReadStream } from "node:fs";
|
||||
import { constants as fsConstants, promises as fs } from "node:fs";
|
||||
import path from "node:path";
|
||||
import { createWorkspaceManifest, WorkspaceManifestMap, workspacePathMatcher, type PathManifest, type WorkspacePaths, type WorkspaceManifestWriter } from "./workspace-manifest.js";
|
||||
import { shouldExcludePath } from "./exclude-patterns.js";
|
||||
import { resolvePaperclipInstanceRootForAdapter } from "./server-utils.js";
|
||||
|
||||
@@ -12,15 +13,44 @@ export type SnapshotEntry =
|
||||
|
||||
export interface DirectorySnapshot {
|
||||
exclude: string[];
|
||||
entries: Map<string, SnapshotEntry>;
|
||||
entries: Map<string, SnapshotEntry> | WorkspaceManifestMap<SnapshotEntry>;
|
||||
ignoredPaths?: WorkspacePaths;
|
||||
}
|
||||
|
||||
export interface SerializedDirectorySnapshot {
|
||||
export interface LegacySerializedDirectorySnapshot {
|
||||
version: 1;
|
||||
exclude: string[];
|
||||
entries: Array<[string, SnapshotEntry]>;
|
||||
}
|
||||
|
||||
export type SerializedDirectorySnapshot = LegacySerializedDirectorySnapshot | {
|
||||
version: 2;
|
||||
exclude: string[];
|
||||
entries: PathManifest;
|
||||
ignoredPaths?: WorkspacePaths;
|
||||
};
|
||||
const ownedDirectorySnapshots = new WeakMap<DirectorySnapshot, string>();
|
||||
export async function disposeDirectorySnapshot(snapshot: DirectorySnapshot | null): Promise<void> {
|
||||
if (!snapshot) return;
|
||||
if (snapshot.entries instanceof WorkspaceManifestMap) {
|
||||
snapshot.entries.close();
|
||||
const ownedDirectory = ownedDirectorySnapshots.get(snapshot);
|
||||
ownedDirectorySnapshots.delete(snapshot);
|
||||
if (ownedDirectory) await fs.rm(ownedDirectory, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
|
||||
function parseManifestEntry(value: string): SnapshotEntry {
|
||||
const result = parseSnapshotEntry(JSON.parse(value));
|
||||
if (!result) throw new Error("Invalid workspace baseline entry");
|
||||
return result;
|
||||
}
|
||||
|
||||
/** Call only after the controller validates a persisted manifest's path and digest. */
|
||||
export function openDirectorySnapshot(value: Extract<SerializedDirectorySnapshot, { version: 2 }>): DirectorySnapshot {
|
||||
return { exclude: value.exclude, entries: new WorkspaceManifestMap(value.entries, parseManifestEntry), ignoredPaths: value.ignoredPaths };
|
||||
}
|
||||
|
||||
function isSafeSnapshotRelativePath(value: string): boolean {
|
||||
if (!value || path.posix.isAbsolute(value) || path.win32.isAbsolute(value)) {
|
||||
return false;
|
||||
@@ -51,6 +81,10 @@ function parseSnapshotEntry(value: unknown): SnapshotEntry | null {
|
||||
export function serializeDirectorySnapshot(
|
||||
snapshot: DirectorySnapshot,
|
||||
): SerializedDirectorySnapshot {
|
||||
if (snapshot.entries instanceof WorkspaceManifestMap) return {
|
||||
version: 2, exclude: [...snapshot.exclude], entries: snapshot.entries.manifest,
|
||||
...(snapshot.ignoredPaths ? { ignoredPaths: snapshot.ignoredPaths } : {}),
|
||||
};
|
||||
return {
|
||||
version: 1,
|
||||
exclude: [...snapshot.exclude],
|
||||
@@ -91,9 +125,11 @@ export function parseDirectorySnapshot(
|
||||
}
|
||||
|
||||
export function directorySnapshotSha256(snapshot: DirectorySnapshot): string {
|
||||
return createHash("sha256")
|
||||
.update(JSON.stringify(serializeDirectorySnapshot(snapshot)))
|
||||
.digest("hex");
|
||||
if (!(snapshot.entries instanceof WorkspaceManifestMap)) return createHash("sha256")
|
||||
.update(JSON.stringify(serializeDirectorySnapshot(snapshot))).digest("hex");
|
||||
const digest = createHash("sha256").update("workspace-baseline-v2\0").update(JSON.stringify(snapshot.exclude));
|
||||
for (const entry of snapshot.entries) digest.update(JSON.stringify(entry)).update("\0");
|
||||
return digest.digest("hex");
|
||||
}
|
||||
|
||||
async function hashFile(filePath: string): Promise<string> {
|
||||
@@ -106,48 +142,24 @@ async function hashFile(filePath: string): Promise<string> {
|
||||
});
|
||||
}
|
||||
|
||||
async function walkDirectory(
|
||||
root: string,
|
||||
exclude: readonly string[],
|
||||
relative = "",
|
||||
out: Map<string, SnapshotEntry> = new Map(),
|
||||
): Promise<Map<string, SnapshotEntry>> {
|
||||
async function* walkDirectory(
|
||||
root: string, exclude: readonly string[], ignored: ReturnType<typeof workspacePathMatcher>, relative = "",
|
||||
): AsyncGenerator<[string, SnapshotEntry]> {
|
||||
const current = relative ? path.join(root, relative) : root;
|
||||
const entries = await fs.readdir(current, { withFileTypes: true }).catch(() => []);
|
||||
entries.sort((left, right) => left.name.localeCompare(right.name));
|
||||
|
||||
for (const entry of entries) {
|
||||
for await (const entry of await fs.opendir(current)) {
|
||||
const nextRelative = relative ? path.posix.join(relative, entry.name) : entry.name;
|
||||
if (shouldExcludePath(nextRelative, exclude)) continue;
|
||||
|
||||
if (shouldExcludePath(nextRelative, exclude) || ignored.matches(nextRelative)) continue;
|
||||
const fullPath = path.join(root, nextRelative);
|
||||
const stats = await fs.lstat(fullPath);
|
||||
if (!stats.isDirectory() && !stats.isSymbolicLink() && !stats.isFile()) {
|
||||
continue;
|
||||
}
|
||||
|
||||
if (stats.isDirectory()) {
|
||||
out.set(nextRelative, { kind: "dir" });
|
||||
await walkDirectory(root, exclude, nextRelative, out);
|
||||
continue;
|
||||
yield [nextRelative, { kind: "dir" }];
|
||||
yield* walkDirectory(root, exclude, ignored, nextRelative);
|
||||
} else if (stats.isSymbolicLink()) {
|
||||
yield [nextRelative, { kind: "symlink", target: await fs.readlink(fullPath) }];
|
||||
} else if (stats.isFile()) {
|
||||
yield [nextRelative, { kind: "file", mode: stats.mode, hash: await hashFile(fullPath) }];
|
||||
}
|
||||
|
||||
if (stats.isSymbolicLink()) {
|
||||
out.set(nextRelative, {
|
||||
kind: "symlink",
|
||||
target: await fs.readlink(fullPath),
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
out.set(nextRelative, {
|
||||
kind: "file",
|
||||
mode: stats.mode,
|
||||
hash: await hashFile(fullPath),
|
||||
});
|
||||
}
|
||||
|
||||
return out;
|
||||
}
|
||||
|
||||
async function readSnapshotEntry(root: string, relative: string): Promise<SnapshotEntry | null> {
|
||||
@@ -444,13 +456,59 @@ async function copySnapshotEntry(sourceDir: string, targetDir: string, relative:
|
||||
|
||||
export async function captureDirectorySnapshot(
|
||||
rootDir: string,
|
||||
options: { exclude?: string[] } = {},
|
||||
options: { exclude?: string[]; ignoredPaths?: WorkspacePaths; diskBacked?: boolean } = {},
|
||||
): Promise<DirectorySnapshot> {
|
||||
const exclude = [...new Set(options.exclude ?? [])];
|
||||
return {
|
||||
exclude,
|
||||
entries: await walkDirectory(rootDir, exclude),
|
||||
};
|
||||
const ignored = workspacePathMatcher(options.ignoredPaths);
|
||||
let writer: WorkspaceManifestWriter | null = null;
|
||||
try {
|
||||
writer = options.diskBacked ? await createWorkspaceManifest("paperclip-workspace-baseline-") : null;
|
||||
const memory = new Map<string, SnapshotEntry>();
|
||||
for await (const [relative, entry] of walkDirectory(rootDir, exclude, ignored)) {
|
||||
if (writer) writer.add("baseline", relative, JSON.stringify(entry));
|
||||
else memory.set(relative, entry);
|
||||
}
|
||||
const manifest = writer?.paths("baseline");
|
||||
writer?.close();
|
||||
const snapshot: DirectorySnapshot = {
|
||||
exclude, ignoredPaths: options.ignoredPaths,
|
||||
entries: manifest ? new WorkspaceManifestMap(manifest, parseManifestEntry) : memory,
|
||||
};
|
||||
if (writer) ownedDirectorySnapshots.set(snapshot, path.dirname(writer.filePath));
|
||||
return snapshot;
|
||||
} catch (error) {
|
||||
writer?.close(false);
|
||||
if (writer) await fs.rm(path.dirname(writer.filePath), { recursive: true, force: true });
|
||||
throw error;
|
||||
} finally { ignored.close(); }
|
||||
}
|
||||
|
||||
/** A disk-backed subset for independent nested-repository merges. */
|
||||
export async function selectDirectorySnapshot(snapshot: DirectorySnapshot, options: {
|
||||
prefix?: string; omit?: string[]; exclude: string[]; ignoredPaths?: WorkspacePaths;
|
||||
}): Promise<DirectorySnapshot> {
|
||||
const writer = await createWorkspaceManifest("paperclip-workspace-baseline-");
|
||||
try {
|
||||
for (const [relative, entry] of snapshot.entries) {
|
||||
if (options.prefix && !relative.startsWith(options.prefix)) continue;
|
||||
if (options.omit?.some((omit) => relative === omit || relative.startsWith(`${omit}/`))) continue;
|
||||
writer.add("baseline", options.prefix ? relative.slice(options.prefix.length) : relative, JSON.stringify(entry));
|
||||
}
|
||||
const result: DirectorySnapshot = { exclude: options.exclude, ignoredPaths: options.ignoredPaths,
|
||||
entries: new WorkspaceManifestMap(writer.paths("baseline"), parseManifestEntry) };
|
||||
writer.close();
|
||||
ownedDirectorySnapshots.set(result, path.dirname(writer.filePath));
|
||||
return result;
|
||||
} catch (error) {
|
||||
writer.close(false);
|
||||
await fs.rm(path.dirname(writer.filePath), { recursive: true, force: true });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
function orderedEntries(snapshot: DirectorySnapshot, reverse = false): Iterable<[string, SnapshotEntry]> {
|
||||
if (snapshot.entries instanceof WorkspaceManifestMap) return snapshot.entries.entries(reverse);
|
||||
return [...snapshot.entries].sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0) * (reverse ? -1 : 1));
|
||||
}
|
||||
|
||||
export async function mergeDirectoryWithBaseline(input: {
|
||||
@@ -460,37 +518,31 @@ export async function mergeDirectoryWithBaseline(input: {
|
||||
beforeApply?: () => Promise<void>;
|
||||
afterApply?: () => Promise<void>;
|
||||
}): Promise<void> {
|
||||
const source = await captureDirectorySnapshot(input.sourceDir, { exclude: input.baseline.exclude });
|
||||
await withDirectoryMergeLock(input.targetDir, async (canonicalTargetDir) => {
|
||||
await input.beforeApply?.();
|
||||
const current = await captureDirectorySnapshot(canonicalTargetDir, { exclude: input.baseline.exclude });
|
||||
const deletedLeafEntries = [...input.baseline.entries.entries()]
|
||||
.filter(([relative, entry]) => entry.kind !== "dir" && !source.entries.has(relative))
|
||||
.sort(([left], [right]) => right.length - left.length);
|
||||
|
||||
for (const [relative, baselineEntry] of deletedLeafEntries) {
|
||||
if (!entriesMatch(current.entries.get(relative), baselineEntry)) continue;
|
||||
await fs.rm(path.join(canonicalTargetDir, relative), { recursive: true, force: true }).catch(() => undefined);
|
||||
}
|
||||
|
||||
const deletedDirs = [...input.baseline.entries.entries()]
|
||||
.filter(([relative, entry]) => entry.kind === "dir" && !source.entries.has(relative))
|
||||
.sort(([left], [right]) => right.length - left.length);
|
||||
|
||||
for (const [relative] of deletedDirs) {
|
||||
await fs.rmdir(path.join(canonicalTargetDir, relative)).catch(() => undefined);
|
||||
}
|
||||
|
||||
const changedSourceEntries = [...source.entries.entries()]
|
||||
.filter(([relative, entry]) => !entriesMatch(input.baseline.entries.get(relative), entry))
|
||||
.sort(([left], [right]) => left.localeCompare(right));
|
||||
|
||||
for (const [relative, entry] of changedSourceEntries) {
|
||||
await copySnapshotEntry(input.sourceDir, canonicalTargetDir, relative, entry);
|
||||
}
|
||||
|
||||
await input.afterApply?.();
|
||||
});
|
||||
const options = { exclude: input.baseline.exclude, ignoredPaths: input.baseline.ignoredPaths, diskBacked: true };
|
||||
const source = await captureDirectorySnapshot(input.sourceDir, options);
|
||||
try {
|
||||
await withDirectoryMergeLock(input.targetDir, async (canonicalTargetDir) => {
|
||||
await input.beforeApply?.();
|
||||
const current = await captureDirectorySnapshot(canonicalTargetDir, options);
|
||||
try {
|
||||
for (const [relative, baselineEntry] of orderedEntries(input.baseline)) {
|
||||
if (baselineEntry.kind === "dir" || source.entries.has(relative)) continue;
|
||||
if (!entriesMatch(current.entries.get(relative), baselineEntry)) continue;
|
||||
await fs.rm(path.join(canonicalTargetDir, relative), { recursive: true, force: true });
|
||||
}
|
||||
// Reverse path order visits descendants before their parent directory.
|
||||
for (const [relative, entry] of orderedEntries(input.baseline, true)) {
|
||||
if (entry.kind === "dir" && !source.entries.has(relative)) await fs.rmdir(path.join(canonicalTargetDir, relative)).catch((error: NodeJS.ErrnoException) => {
|
||||
if (error.code !== "ENOENT" && error.code !== "ENOTEMPTY") throw error;
|
||||
});
|
||||
}
|
||||
for (const [relative, entry] of orderedEntries(source)) {
|
||||
if (!entriesMatch(input.baseline.entries.get(relative), entry)) await copySnapshotEntry(input.sourceDir, canonicalTargetDir, relative, entry);
|
||||
}
|
||||
await input.afterApply?.();
|
||||
} finally { await disposeDirectorySnapshot(current); }
|
||||
});
|
||||
} finally { await disposeDirectorySnapshot(source); }
|
||||
}
|
||||
|
||||
export async function directoryEntryMatchesBaseline(
|
||||
|
||||
@@ -1,10 +1,12 @@
|
||||
import { mkdtemp, readdir, rm } from "node:fs/promises";
|
||||
import { mkdtemp, readdir, rm, mkdir, writeFile, readFile, appendFile, symlink, stat } from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
directorySnapshotSha256,
|
||||
serializeDirectorySnapshot,
|
||||
captureDirectorySnapshot,
|
||||
disposeDirectorySnapshot,
|
||||
} from "@paperclipai/adapter-utils/workspace-restore-merge";
|
||||
|
||||
import {
|
||||
@@ -13,6 +15,7 @@ import {
|
||||
readNativeWorkspaceSyncReference,
|
||||
resumeNativeWorkspaceSync,
|
||||
prepareNativeWorkspaceSync,
|
||||
cleanupNativeWorkspaceSync,
|
||||
} from "../services/native-runtime/native-workspace-sync.js";
|
||||
|
||||
const digest = "a".repeat(64);
|
||||
@@ -149,6 +152,55 @@ describe("native workspace sync durable metadata", () => {
|
||||
).toThrow("native_workspace_sync_descriptor_digest_invalid");
|
||||
});
|
||||
|
||||
it.each(["corrupt", "symlink", "foreign"] as const)("persists compact v2 manifests and rejects %s recovery storage", async (tamper) => {
|
||||
const paperclipHome = await mkdtemp(path.join(os.tmpdir(), "paperclip-native-manifest-"));
|
||||
cleanupDirs.push(paperclipHome);
|
||||
process.env.PAPERCLIP_HOME = paperclipHome;
|
||||
process.env.PAPERCLIP_INSTANCE_ID = "manifest-test";
|
||||
const workspace = path.join(paperclipHome, "workspace");
|
||||
await mkdir(workspace);
|
||||
await writeFile(path.join(workspace, "private-filename-雪"), "baseline");
|
||||
const baseline = await captureDirectorySnapshot(workspace, { diskBacked: true });
|
||||
const initial = serializeDirectorySnapshot(baseline);
|
||||
if (initial.version !== 2) throw new Error("Expected a disk baseline");
|
||||
const scratch = initial.entries.filePath;
|
||||
const runId = "manifest-run";
|
||||
const stateDir = path.dirname(nativeWorkspaceSyncInternals.durableSeedPaths(runId).workspaceArchivePath);
|
||||
await mkdir(stateDir, { recursive: true });
|
||||
await nativeWorkspaceSyncInternals.persistSnapshotManifests(runId, { baseline, gitSnapshot: null });
|
||||
await expect(stat(scratch)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
const serialized = serializeDirectorySnapshot(baseline);
|
||||
if (serialized.version !== 2) throw new Error("Expected v2");
|
||||
const descriptor = {
|
||||
schema: "paperclip.native-workspace-sync/v2" as const,
|
||||
binding: { runId, companyId: "company", workspaceId: "workspace", leaseId: "lease", providerLeaseId: "sandbox", localCwd: workspace, remoteCwd: "/workspace" },
|
||||
state: "prepared" as const, baselineSha256: directorySnapshotSha256(baseline),
|
||||
baseline: serialized, gitSnapshot: null, seed: null,
|
||||
createdAt: new Date().toISOString(), finalizedAt: null, finalHostSha256: null, resourceDisposition: null,
|
||||
};
|
||||
let reference = await nativeWorkspaceSyncInternals.writeDescriptor(descriptor);
|
||||
const raw = await readFile(nativeWorkspaceSyncInternals.descriptorPath(runId, reference.descriptorSha256), "utf8");
|
||||
expect(raw.length).toBeLessThan(4096);
|
||||
expect(raw).not.toContain("private-filename");
|
||||
const recovered = await nativeWorkspaceSyncInternals.readDescriptor({ runId, reference });
|
||||
expect(recovered.baseline.entries.get("private-filename-雪")).toMatchObject({ kind: "file" });
|
||||
await disposeDirectorySnapshot(recovered.baseline);
|
||||
if (tamper === "corrupt") {
|
||||
await appendFile(serialized.entries.filePath, "corrupt");
|
||||
} else if (tamper === "symlink") {
|
||||
const copy = path.join(paperclipHome, "copied.sqlite");
|
||||
await writeFile(copy, await readFile(serialized.entries.filePath));
|
||||
await rm(serialized.entries.filePath);
|
||||
await symlink(copy, serialized.entries.filePath);
|
||||
} else {
|
||||
descriptor.baseline.entries.filePath = path.join(paperclipHome, "foreign.sqlite");
|
||||
reference = await nativeWorkspaceSyncInternals.writeDescriptor(descriptor);
|
||||
}
|
||||
await expect(nativeWorkspaceSyncInternals.readDescriptor({ runId, reference })).rejects.toThrow();
|
||||
await cleanupNativeWorkspaceSync(runId);
|
||||
await expect(stat(stateDir)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
});
|
||||
|
||||
it.each([false, true])("writes one immutable descriptor when the same state is replayed (multiple repositories: %s)", async (multipleRepositories) => {
|
||||
const paperclipHome = await mkdtemp(
|
||||
path.join(os.tmpdir(), "paperclip-native-workspace-sync-"),
|
||||
|
||||
@@ -39,9 +39,9 @@ import { buildHeartbeatRunStatusLiveEventPayload } from "./heartbeat-run-status-
|
||||
export { buildHeartbeatRunStatusLiveEventPayload } from "./heartbeat-run-status-payload.js";
|
||||
import { buildExecutionContinuation, StaleExecutionContinuationError } from "./execution-continuation.js";
|
||||
import { renderPaperclipWakePrompt } from "@paperclipai/adapter-utils/server-utils";
|
||||
import { PROJECT_REPOSITORIES_DIR, readGitWorkspaceSnapshot } from "@paperclipai/adapter-utils/git-workspace-sync";
|
||||
import { PROJECT_REPOSITORIES_DIR, readGitWorkspaceSnapshot, disposeGitWorkspaceSnapshot } from "@paperclipai/adapter-utils/git-workspace-sync";
|
||||
import { isWorkspaceGitScanError, WorkspaceGitScanError, WORKSPACE_GIT_SCAN_ERROR_CODES } from "./workspace-git-operation-scheduler.js";
|
||||
import { captureDirectorySnapshot, mergeDirectoryWithBaseline } from "@paperclipai/adapter-utils/workspace-restore-merge";
|
||||
import { captureDirectorySnapshot, disposeDirectorySnapshot, mergeDirectoryWithBaseline } from "@paperclipai/adapter-utils/workspace-restore-merge";
|
||||
import { initializeRunIdentity, explicitOperatorRunIdentity } from "./run-identity.js";
|
||||
import {
|
||||
assertDurableChatWakeupReceipt,
|
||||
@@ -2516,8 +2516,14 @@ async function materializeManagedProjectWorkspace(
|
||||
if (input.localSource) {
|
||||
const snapshot = await readGitWorkspaceSnapshot(input.localSource, false);
|
||||
if (!snapshot) throw new Error("Configured repository folder is not a Git checkout");
|
||||
const baseline = await captureDirectorySnapshot(cloneTmpDir, { exclude: [".git", ".paperclip-runtime", PROJECT_REPOSITORIES_DIR, ...snapshot.ignoredPaths] });
|
||||
await mergeDirectoryWithBaseline({ baseline, sourceDir: input.localSource, targetDir: cloneTmpDir });
|
||||
let baseline;
|
||||
try {
|
||||
baseline = await captureDirectorySnapshot(cloneTmpDir, { exclude: [".git", ".paperclip-runtime", PROJECT_REPOSITORIES_DIR], ignoredPaths: snapshot.ignoredPaths, diskBacked: true });
|
||||
await mergeDirectoryWithBaseline({ baseline, sourceDir: input.localSource, targetDir: cloneTmpDir });
|
||||
} finally {
|
||||
if (baseline) await disposeDirectorySnapshot(baseline);
|
||||
await disposeGitWorkspaceSnapshot(snapshot);
|
||||
}
|
||||
await execFile("git", ["-C", cloneTmpDir, "remote", "set-url", "origin", input.repoUrl], { timeout: 10_000 });
|
||||
} else if (input.repoRef) {
|
||||
await execFile("git", ["-C", cloneTmpDir, "checkout", input.repoRef], { timeout: MANAGED_WORKSPACE_GIT_CLONE_TIMEOUT_MS });
|
||||
|
||||
@@ -11,9 +11,11 @@ import {
|
||||
type AdapterExecutionTarget,
|
||||
type PreparedAdapterExecutionTargetRuntime,
|
||||
} from "@paperclipai/adapter-utils/execution-target";
|
||||
import type { GitWorkspaceSnapshot } from "@paperclipai/adapter-utils/git-workspace-sync";
|
||||
import { disposeGitWorkspaceSnapshot, type GitWorkspaceSnapshot } from "@paperclipai/adapter-utils/git-workspace-sync";
|
||||
import {
|
||||
directorySnapshotSha256,
|
||||
disposeDirectorySnapshot,
|
||||
openDirectorySnapshot,
|
||||
parseDirectorySnapshot,
|
||||
serializeDirectorySnapshot,
|
||||
type DirectorySnapshot,
|
||||
@@ -23,11 +25,14 @@ import type {
|
||||
WorkspaceDurableSeedPaths,
|
||||
WorkspaceInboundMode,
|
||||
} from "@paperclipai/adapter-utils/sandbox-managed-runtime";
|
||||
import { isPathManifest, manifestFileSha256, readManifestRecords, type PathManifest, type WorkspacePaths } from "@paperclipai/adapter-utils/workspace-manifest";
|
||||
import { resolvePaperclipInstanceRoot } from "../../home-paths.js";
|
||||
import { parseObject } from "../../adapters/utils.js";
|
||||
import type { NativeRestartRecoveryClaim } from "./native-restart-recovery.js";
|
||||
|
||||
const DESCRIPTOR_SCHEMA = "paperclip.native-workspace-sync/v1";
|
||||
const LEGACY_DESCRIPTOR_SCHEMA = "paperclip.native-workspace-sync/v1";
|
||||
const DESCRIPTOR_SCHEMA = "paperclip.native-workspace-sync/v2";
|
||||
type DescriptorSchema = typeof DESCRIPTOR_SCHEMA | typeof LEGACY_DESCRIPTOR_SCHEMA;
|
||||
const STAMP_SCHEMA = "paperclip.native-workspace-stamp/v1";
|
||||
const STATE_ROOT_NAME = "native-workspace-sync";
|
||||
const DESCRIPTOR_NAME = "descriptor";
|
||||
@@ -42,7 +47,7 @@ export type NativeWorkspaceResourceDisposition =
|
||||
"keep_running" | "stop_and_retain" | "destroy";
|
||||
|
||||
interface NativeWorkspaceSyncDescriptor {
|
||||
schema: typeof DESCRIPTOR_SCHEMA;
|
||||
schema: DescriptorSchema;
|
||||
binding: {
|
||||
runId: string;
|
||||
companyId: string;
|
||||
@@ -67,7 +72,7 @@ interface NativeWorkspaceSyncDescriptor {
|
||||
}
|
||||
|
||||
export interface NativeWorkspaceSyncReference {
|
||||
schema: typeof DESCRIPTOR_SCHEMA;
|
||||
schema: DescriptorSchema;
|
||||
state: NativeWorkspaceSyncState;
|
||||
descriptorSha256: string;
|
||||
baselineSha256: string;
|
||||
@@ -258,7 +263,7 @@ async function writeDescriptor(
|
||||
await fs.rm(tempPath, { force: true }).catch(() => undefined);
|
||||
}
|
||||
return {
|
||||
schema: DESCRIPTOR_SCHEMA,
|
||||
schema: descriptor.schema,
|
||||
state: descriptor.state,
|
||||
descriptorSha256,
|
||||
baselineSha256: descriptor.baselineSha256,
|
||||
@@ -276,7 +281,7 @@ export function readNativeWorkspaceSyncReference(
|
||||
): NativeWorkspaceSyncReference | null {
|
||||
const candidate = parseObject(value);
|
||||
if (
|
||||
candidate.schema !== DESCRIPTOR_SCHEMA ||
|
||||
(candidate.schema !== DESCRIPTOR_SCHEMA && candidate.schema !== LEGACY_DESCRIPTOR_SCHEMA) ||
|
||||
(candidate.state !== "prepared" && candidate.state !== "finalized") ||
|
||||
typeof candidate.descriptorSha256 !== "string" ||
|
||||
!/^[0-9a-f]{64}$/.test(candidate.descriptorSha256) ||
|
||||
@@ -299,7 +304,7 @@ export function readNativeWorkspaceSyncReference(
|
||||
return null;
|
||||
}
|
||||
return {
|
||||
schema: DESCRIPTOR_SCHEMA,
|
||||
schema: candidate.schema,
|
||||
state: candidate.state,
|
||||
descriptorSha256: candidate.descriptorSha256,
|
||||
baselineSha256: candidate.baselineSha256,
|
||||
@@ -320,35 +325,76 @@ export function readNativeWorkspaceSyncReference(
|
||||
};
|
||||
}
|
||||
|
||||
function validManifestShape(value: unknown): value is PathManifest {
|
||||
const record = parseObject(value);
|
||||
return record.kind === "path_manifest" && record.version === 1 && typeof record.filePath === "string"
|
||||
&& typeof record.category === "string" && ["overlay", "deleted", "ignored", "baseline"].includes(record.category)
|
||||
&& typeof record.count === "number" && Number.isSafeInteger(record.count) && record.count >= 0;
|
||||
}
|
||||
|
||||
function* gitManifests(snapshot: GitWorkspaceSnapshot | null): Generator<PathManifest> {
|
||||
if (!snapshot) return;
|
||||
for (const paths of [snapshot.overlayPaths, snapshot.deletedPaths, snapshot.ignoredPaths]) if (isPathManifest(paths)) yield paths;
|
||||
for (const repo of snapshot.repositories ?? []) yield* gitManifests(repo.snapshot);
|
||||
}
|
||||
|
||||
async function verifyManifest(runId: string, manifest: PathManifest): Promise<void> {
|
||||
const directory = descriptorDirectory(runId);
|
||||
const name = path.basename(manifest.filePath);
|
||||
const match = /^manifest\.([0-9a-f]{64})\.sqlite$/.exec(name);
|
||||
if (!match || manifest.filePath !== path.join(directory, name)) throw new Error("native_workspace_sync_manifest_path_invalid");
|
||||
for (const dir of [path.dirname(directory), directory]) {
|
||||
const stat = await fs.lstat(dir);
|
||||
if (!stat.isDirectory() || stat.isSymbolicLink()) throw new Error("native_workspace_sync_manifest_root_invalid");
|
||||
}
|
||||
if (await manifestFileSha256(manifest.filePath) !== match[1]) throw new Error("native_workspace_sync_manifest_digest_mismatch");
|
||||
// Verify every record before consumers can mutate a workspace on recovery.
|
||||
for (const _record of readManifestRecords(manifest)) { /* bounded validation */ }
|
||||
}
|
||||
|
||||
async function persistSnapshotManifests(runId: string, snapshot: NonNullable<PreparedAdapterExecutionTargetRuntime["workspaceSyncSnapshot"]>): Promise<void> {
|
||||
const baseline = serializeDirectorySnapshot(snapshot.baseline);
|
||||
const manifests = function* () {
|
||||
if (baseline.version === 2) {
|
||||
yield baseline.entries;
|
||||
if (baseline.ignoredPaths && isPathManifest(baseline.ignoredPaths)) yield baseline.ignoredPaths;
|
||||
}
|
||||
yield* gitManifests(snapshot.gitSnapshot);
|
||||
};
|
||||
const copies = new Map<string, string>();
|
||||
for (const manifest of manifests()) {
|
||||
const oldPath = manifest.filePath;
|
||||
const previous = copies.get(oldPath);
|
||||
if (previous) { manifest.filePath = previous; continue; }
|
||||
// A shared reference may already have been updated by the baseline.
|
||||
if (path.dirname(oldPath) === descriptorDirectory(runId)) continue;
|
||||
const temporary = path.join(descriptorDirectory(runId), `manifest.${randomUUID()}.tmp`);
|
||||
try {
|
||||
await fs.copyFile(oldPath, temporary);
|
||||
await fs.chmod(temporary, 0o600);
|
||||
const digest = await manifestFileSha256(temporary);
|
||||
const target = path.join(descriptorDirectory(runId), `manifest.${digest}.sqlite`);
|
||||
await fs.rename(temporary, target);
|
||||
copies.set(oldPath, target);
|
||||
manifest.filePath = target;
|
||||
} finally { await fs.rm(temporary, { force: true }); }
|
||||
}
|
||||
// Ownership tracks original scratch directories, never the new durable path.
|
||||
await disposeDirectorySnapshot(snapshot.baseline);
|
||||
await disposeGitWorkspaceSnapshot(snapshot.gitSnapshot);
|
||||
}
|
||||
|
||||
function parseGitSnapshot(
|
||||
value: unknown,
|
||||
nested = false,
|
||||
): GitWorkspaceSnapshot | null | undefined {
|
||||
if (value === null) return null;
|
||||
const candidate = parseObject(value);
|
||||
const paths = [
|
||||
candidate.overlayPaths,
|
||||
candidate.deletedPaths,
|
||||
candidate.ignoredPaths,
|
||||
];
|
||||
if (
|
||||
typeof candidate.headCommit !== "string" ||
|
||||
(candidate.branchName !== null &&
|
||||
typeof candidate.branchName !== "string") ||
|
||||
!paths.every(
|
||||
(entries) =>
|
||||
Array.isArray(entries) &&
|
||||
entries.every(
|
||||
(entry) =>
|
||||
typeof entry === "string" &&
|
||||
!path.posix.isAbsolute(entry) &&
|
||||
!path.win32.isAbsolute(entry) &&
|
||||
!entry.split(/[\\/]/).some((segment) => segment === ".."),
|
||||
),
|
||||
)
|
||||
) {
|
||||
return undefined;
|
||||
}
|
||||
const paths = [candidate.overlayPaths, candidate.deletedPaths, candidate.ignoredPaths];
|
||||
if (typeof candidate.headCommit !== "string" || (candidate.branchName !== null && typeof candidate.branchName !== "string")
|
||||
|| !paths.every((entries) => validManifestShape(entries) || (Array.isArray(entries) && entries.every((entry) =>
|
||||
typeof entry === "string" && entry.length > 0 && !entry.includes("\0") && !path.posix.isAbsolute(entry) && !path.win32.isAbsolute(entry)
|
||||
&& !entry.split(/[\\/]/).some((segment) => segment === ".." || segment === "."))))) return undefined;
|
||||
const repositories: NonNullable<GitWorkspaceSnapshot["repositories"]> = [];
|
||||
if (candidate.repositories !== undefined) {
|
||||
if (nested || !Array.isArray(candidate.repositories)) return undefined;
|
||||
@@ -365,9 +411,9 @@ function parseGitSnapshot(
|
||||
return {
|
||||
headCommit: candidate.headCommit,
|
||||
branchName: candidate.branchName as string | null,
|
||||
overlayPaths: [...(candidate.overlayPaths as string[])],
|
||||
deletedPaths: [...(candidate.deletedPaths as string[])],
|
||||
ignoredPaths: [...(candidate.ignoredPaths as string[])],
|
||||
overlayPaths: candidate.overlayPaths as WorkspacePaths,
|
||||
deletedPaths: candidate.deletedPaths as WorkspacePaths,
|
||||
ignoredPaths: candidate.ignoredPaths as WorkspacePaths,
|
||||
...(repositories.length ? { repositories } : {}),
|
||||
};
|
||||
}
|
||||
@@ -413,8 +459,20 @@ async function readDescriptor(input: {
|
||||
const parsed = JSON.parse(body) as unknown;
|
||||
const candidate = parseObject(parsed);
|
||||
const binding = parseObject(candidate.binding);
|
||||
const baseline = parseDirectorySnapshot(candidate.baseline);
|
||||
const rawBaseline = parseObject(candidate.baseline);
|
||||
let baseline: DirectorySnapshot | null;
|
||||
if (candidate.schema === DESCRIPTOR_SCHEMA && rawBaseline.version === 2
|
||||
&& Array.isArray(rawBaseline.exclude) && rawBaseline.exclude.every((entry) => typeof entry === "string")
|
||||
&& validManifestShape(rawBaseline.entries) && rawBaseline.entries.category === "baseline"
|
||||
&& (rawBaseline.ignoredPaths === undefined || validManifestShape(rawBaseline.ignoredPaths))) {
|
||||
await verifyManifest(input.runId, rawBaseline.entries);
|
||||
if (validManifestShape(rawBaseline.ignoredPaths)) await verifyManifest(input.runId, rawBaseline.ignoredPaths);
|
||||
baseline = openDirectorySnapshot(rawBaseline as unknown as Extract<SerializedDirectorySnapshot, { version: 2 }>);
|
||||
} else {
|
||||
baseline = parseDirectorySnapshot(candidate.baseline);
|
||||
}
|
||||
const gitSnapshot = parseGitSnapshot(candidate.gitSnapshot);
|
||||
for (const manifest of gitManifests(gitSnapshot ?? null)) await verifyManifest(input.runId, manifest);
|
||||
const rawSeed =
|
||||
candidate.seed === null || candidate.seed === undefined
|
||||
? null
|
||||
@@ -448,7 +506,7 @@ async function readDescriptor(input: {
|
||||
? null
|
||||
: undefined;
|
||||
if (
|
||||
candidate.schema !== DESCRIPTOR_SCHEMA ||
|
||||
(candidate.schema !== DESCRIPTOR_SCHEMA && candidate.schema !== LEGACY_DESCRIPTOR_SCHEMA) ||
|
||||
(candidate.state !== "prepared" && candidate.state !== "finalized") ||
|
||||
!baseline ||
|
||||
gitSnapshot === undefined ||
|
||||
@@ -468,6 +526,7 @@ async function readDescriptor(input: {
|
||||
directorySnapshotSha256(baseline) !== candidate.baselineSha256 ||
|
||||
typeof candidate.createdAt !== "string" ||
|
||||
!Number.isFinite(Date.parse(candidate.createdAt)) ||
|
||||
candidate.schema !== input.reference.schema ||
|
||||
candidate.state !== input.reference.state ||
|
||||
finalHostSha256 === undefined ||
|
||||
finalHostSha256 !== input.reference.finalHostSha256 ||
|
||||
@@ -484,7 +543,7 @@ async function readDescriptor(input: {
|
||||
}
|
||||
return {
|
||||
descriptor: {
|
||||
schema: DESCRIPTOR_SCHEMA,
|
||||
schema: candidate.schema,
|
||||
binding: {
|
||||
runId: binding.runId as string,
|
||||
companyId: binding.companyId,
|
||||
@@ -717,9 +776,12 @@ async function finalizePreparedRuntime(input: {
|
||||
({ captureDirectorySnapshot }) =>
|
||||
captureDirectorySnapshot(input.descriptor.binding.localCwd, {
|
||||
exclude: input.runtime.workspaceSyncSnapshot?.baseline.exclude ?? [],
|
||||
ignoredPaths: input.runtime.workspaceSyncSnapshot?.baseline.ignoredPaths,
|
||||
diskBacked: true,
|
||||
}),
|
||||
);
|
||||
const finalHostSha256 = directorySnapshotSha256(finalSnapshot);
|
||||
await disposeDirectorySnapshot(finalSnapshot);
|
||||
const stamp = finalizedWorkspaceStamp({
|
||||
descriptor: input.descriptor,
|
||||
hostSha256: finalHostSha256,
|
||||
@@ -776,12 +838,13 @@ export async function prepareNativeWorkspaceSync(input: {
|
||||
parseObject(run.runnerProfileJson).nativeWorkspaceSync,
|
||||
);
|
||||
|
||||
let runtime: PreparedAdapterExecutionTargetRuntime;
|
||||
let runtime: PreparedAdapterExecutionTargetRuntime | undefined;
|
||||
let descriptor: NativeWorkspaceSyncDescriptor;
|
||||
let mode: WorkspaceInboundMode = "host_current";
|
||||
const seedPaths = durableSeedPaths(input.runId);
|
||||
await ensurePrivateDirectory(descriptorDirectory(input.runId));
|
||||
|
||||
try {
|
||||
if (existingReference) {
|
||||
const existing = await readDescriptor({
|
||||
runId: input.runId,
|
||||
@@ -887,6 +950,7 @@ export async function prepareNativeWorkspaceSync(input: {
|
||||
}
|
||||
const snapshot = runtime.workspaceSyncSnapshot;
|
||||
if (!snapshot) throw new Error("native_workspace_sync_snapshot_missing");
|
||||
await persistSnapshotManifests(input.runId, snapshot);
|
||||
const now = new Date().toISOString();
|
||||
const workspaceArchiveSha256 = await sha256File(
|
||||
seedPaths.workspaceArchivePath,
|
||||
@@ -917,6 +981,7 @@ export async function prepareNativeWorkspaceSync(input: {
|
||||
};
|
||||
}
|
||||
|
||||
const preparedRuntime = runtime;
|
||||
let reference = await writeDescriptor(descriptor);
|
||||
await persistRunReference(input.db, input.runId, reference);
|
||||
let restorePromise: Promise<void> | null = null;
|
||||
@@ -931,7 +996,7 @@ export async function prepareNativeWorkspaceSync(input: {
|
||||
db: input.db,
|
||||
runId: input.runId,
|
||||
target,
|
||||
runtime,
|
||||
runtime: preparedRuntime,
|
||||
descriptor,
|
||||
})
|
||||
.then((finalizedReference) => {
|
||||
@@ -944,8 +1009,13 @@ export async function prepareNativeWorkspaceSync(input: {
|
||||
}
|
||||
await restorePromise;
|
||||
},
|
||||
cleanup: () => cleanupNativeWorkspaceSync(input.runId),
|
||||
cleanup: async () => { await preparedRuntime.cleanupWorkspaceSnapshot?.(); await cleanupNativeWorkspaceSync(input.runId); },
|
||||
};
|
||||
} catch (error) {
|
||||
await runtime?.cleanupWorkspaceSnapshot?.();
|
||||
if (!existingReference) await cleanupNativeWorkspaceSync(input.runId);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
export async function resumeNativeWorkspaceSync(input: {
|
||||
@@ -1022,6 +1092,7 @@ export const nativeWorkspaceSyncInternals = {
|
||||
descriptorPath,
|
||||
legacyDescriptorPath,
|
||||
durableSeedPaths,
|
||||
persistSnapshotManifests,
|
||||
writeDescriptor,
|
||||
readDescriptor,
|
||||
readReference: readNativeWorkspaceSyncReference,
|
||||
|
||||
@@ -46,6 +46,55 @@ afterEach(async () => {
|
||||
await Promise.all(tempPaths.splice(0).map((tempPath) => fs.rm(tempPath, { recursive: true, force: true })));
|
||||
});
|
||||
|
||||
it("streams with backpressure and waits for an active sink before releasing cancellation", async () => {
|
||||
const workspace = await makeWorkspace();
|
||||
const entered = deferred();
|
||||
const release = deferred();
|
||||
const scheduler = createWorkspaceGitOperationScheduler({
|
||||
gitBinary: process.execPath,
|
||||
gitArgsPrefix: ["-e", "process.stdout.write(Buffer.alloc(1024*1024)); setInterval(()=>{},1000)", "--"],
|
||||
concurrency: 1,
|
||||
killGraceMs: 10,
|
||||
});
|
||||
const controller = new AbortController();
|
||||
let settled = false;
|
||||
const pending = scheduler.run({
|
||||
...scanInput(workspace, "stream"), signal: controller.signal, maxStdoutBytes: 1,
|
||||
onStdout: async (chunk) => {
|
||||
expect(chunk.length).toBeLessThanOrEqual(64 * 1024);
|
||||
entered.resolve();
|
||||
await release.promise;
|
||||
},
|
||||
});
|
||||
const observed = pending.catch((error) => { settled = true; return error; });
|
||||
await entered.promise;
|
||||
controller.abort();
|
||||
await new Promise((resolve) => setTimeout(resolve, 30));
|
||||
expect(settled).toBe(false);
|
||||
expect(scheduler.snapshot().activeCount).toBe(1);
|
||||
release.resolve();
|
||||
expect(await observed).toMatchObject({ code: WORKSPACE_GIT_SCAN_ERROR_CODES.cancelled });
|
||||
expect(scheduler.snapshot()).toMatchObject({ activeCount: 0, inFlightCount: 0 });
|
||||
});
|
||||
|
||||
it("does not coalesce or cache caller-owned streaming sinks and fails on a sink error", async () => {
|
||||
const workspace = await makeWorkspace();
|
||||
const scheduler = createWorkspaceGitOperationScheduler({
|
||||
gitBinary: process.execPath, gitArgsPrefix: ["-e", "process.stdout.write('path\\0')", "--"], concurrency: 1,
|
||||
});
|
||||
const bytes = [0, 0];
|
||||
await Promise.all(bytes.map((_, index) => scheduler.run({
|
||||
...scanInput(workspace, "same"), cacheTtlMs: 10000, maxStdoutBytes: 1,
|
||||
onStdout: (chunk) => { bytes[index] += chunk.length; },
|
||||
})));
|
||||
expect(bytes).toEqual([5, 5]);
|
||||
expect(scheduler.snapshot()).toMatchObject({ cacheEntryCount: 0, totals: { started: 2, singleFlightJoins: 0 } });
|
||||
await expect(scheduler.run({
|
||||
...scanInput(workspace, "same"), onStdout: () => { throw new Error("disk full"); },
|
||||
})).rejects.toMatchObject({ code: WORKSPACE_GIT_SCAN_ERROR_CODES.failed });
|
||||
expect(scheduler.snapshot()).toMatchObject({ activeCount: 0, inFlightCount: 0 });
|
||||
});
|
||||
|
||||
describe("WorkspaceGitOperationScheduler", () => {
|
||||
it("loads bounded process defaults and overrides from the environment", () => {
|
||||
expect(workspaceGitSchedulerOptionsFromEnv({})).toEqual({
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import { spawn, type ChildProcess } from "node:child_process";
|
||||
import { createHash } from "node:crypto";
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import fs from "node:fs/promises";
|
||||
import { setExpensiveWorkspaceGitExecutor } from "@paperclipai/adapter-utils/git-workspace-sync";
|
||||
import { runWorkspaceGitProcess } from "@paperclipai/adapter-utils/workspace-git-stream";
|
||||
import { HttpError } from "../errors.js";
|
||||
import { logger } from "../middleware/logger.js";
|
||||
|
||||
@@ -53,6 +53,8 @@ export interface WorkspaceGitScanResult {
|
||||
}
|
||||
|
||||
export interface WorkspaceGitScanInput {
|
||||
/** A backpressured, non-cacheable stdout consumer. Streaming jobs never coalesce. */
|
||||
onStdout?: (chunk: Buffer) => Promise<void> | void;
|
||||
workspacePath: string;
|
||||
args: readonly string[];
|
||||
operation: string;
|
||||
@@ -91,6 +93,8 @@ export interface WorkspaceGitSchedulerSnapshot {
|
||||
}
|
||||
|
||||
export interface WorkspaceGitRunnerInput {
|
||||
/** A backpressured, non-cacheable stdout consumer. Streaming jobs never coalesce. */
|
||||
onStdout?: (chunk: Buffer) => Promise<void> | void;
|
||||
canonicalWorkspacePath: string;
|
||||
args: readonly string[];
|
||||
env?: NodeJS.ProcessEnv;
|
||||
@@ -137,6 +141,8 @@ interface Waiter {
|
||||
}
|
||||
|
||||
interface PendingScan {
|
||||
/** A backpressured, non-cacheable stdout consumer. Streaming jobs never coalesce. */
|
||||
onStdout?: (chunk: Buffer) => Promise<void> | void;
|
||||
key: string;
|
||||
operation: string;
|
||||
canonicalWorkspacePath: string;
|
||||
@@ -254,151 +260,20 @@ function abortError(workspaceHash: string): WorkspaceGitScanError {
|
||||
);
|
||||
}
|
||||
|
||||
function signalChild(child: ChildProcess, signal: NodeJS.Signals): void {
|
||||
if (child.exitCode !== null || child.signalCode !== null) return;
|
||||
if (process.platform !== "win32" && child.pid) {
|
||||
function createSpawnRunner(input: { gitBinary: string; gitArgsPrefix: readonly string[] }): WorkspaceGitRunner {
|
||||
return async (runInput) => {
|
||||
try {
|
||||
process.kill(-child.pid, signal);
|
||||
return;
|
||||
} catch {
|
||||
// Fall back to the direct child if the process group already disappeared.
|
||||
return await runWorkspaceGitProcess({ ...runInput, ...input, cwd: runInput.canonicalWorkspacePath });
|
||||
} catch (error) {
|
||||
const candidate = error as { code?: WorkspaceGitScanErrorCode; details?: Record<string, unknown> };
|
||||
throw new WorkspaceGitScanError(
|
||||
candidate.code && Object.values(WORKSPACE_GIT_SCAN_ERROR_CODES).includes(candidate.code)
|
||||
? candidate.code : WORKSPACE_GIT_SCAN_ERROR_CODES.failed,
|
||||
error instanceof Error ? error.message : "Workspace Git scan failed",
|
||||
{ ...candidate.details, workspaceHash: workspaceIdentity(runInput.canonicalWorkspacePath) },
|
||||
);
|
||||
}
|
||||
}
|
||||
try {
|
||||
child.kill(signal);
|
||||
} catch {
|
||||
// The close/error handler owns settlement; an already-dead child is benign.
|
||||
}
|
||||
}
|
||||
|
||||
function createSpawnRunner(input: {
|
||||
gitBinary: string;
|
||||
gitArgsPrefix: readonly string[];
|
||||
}): WorkspaceGitRunner {
|
||||
return (runInput) => new Promise<WorkspaceGitRunnerResult>((resolve, reject) => {
|
||||
if (runInput.signal.aborted) {
|
||||
reject(abortError(workspaceIdentity(runInput.canonicalWorkspacePath)));
|
||||
return;
|
||||
}
|
||||
|
||||
const child = spawn(
|
||||
input.gitBinary,
|
||||
[...input.gitArgsPrefix, "-C", runInput.canonicalWorkspacePath, ...runInput.args],
|
||||
{
|
||||
cwd: runInput.canonicalWorkspacePath,
|
||||
env: runInput.env ?? process.env,
|
||||
stdio: ["ignore", "pipe", "pipe"],
|
||||
detached: process.platform !== "win32",
|
||||
windowsHide: true,
|
||||
},
|
||||
);
|
||||
const stdoutChunks: Buffer[] = [];
|
||||
const stderrChunks: Buffer[] = [];
|
||||
let stdoutBytes = 0;
|
||||
let stderrBytes = 0;
|
||||
let termination: "timeout" | "cancelled" | "output_limit" | null = null;
|
||||
let spawnError: Error | null = null;
|
||||
let settled = false;
|
||||
|
||||
const terminate = (reason: NonNullable<typeof termination>) => {
|
||||
if (termination) return;
|
||||
termination = reason;
|
||||
signalChild(child, "SIGTERM");
|
||||
killTimer = setTimeout(() => signalChild(child, "SIGKILL"), runInput.killGraceMs);
|
||||
killTimer.unref?.();
|
||||
};
|
||||
|
||||
const append = (
|
||||
chunk: Buffer | string,
|
||||
chunks: Buffer[],
|
||||
currentBytes: number,
|
||||
maxBytes: number,
|
||||
): number => {
|
||||
const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
|
||||
const remaining = Math.max(0, maxBytes - currentBytes);
|
||||
if (remaining > 0) chunks.push(buffer.subarray(0, remaining));
|
||||
const nextBytes = currentBytes + buffer.length;
|
||||
if (nextBytes > maxBytes) terminate("output_limit");
|
||||
return nextBytes;
|
||||
};
|
||||
|
||||
const onAbort = () => terminate("cancelled");
|
||||
runInput.signal.addEventListener("abort", onAbort, { once: true });
|
||||
child.stdout?.on("data", (chunk) => {
|
||||
stdoutBytes = append(chunk, stdoutChunks, stdoutBytes, runInput.maxStdoutBytes);
|
||||
});
|
||||
child.stderr?.on("data", (chunk) => {
|
||||
stderrBytes = append(chunk, stderrChunks, stderrBytes, runInput.maxStderrBytes);
|
||||
});
|
||||
child.once("error", (error) => {
|
||||
spawnError = error;
|
||||
});
|
||||
|
||||
const timeoutTimer = setTimeout(() => terminate("timeout"), runInput.timeoutMs);
|
||||
timeoutTimer.unref?.();
|
||||
let killTimer: NodeJS.Timeout | null = null;
|
||||
|
||||
child.once("close", (code, childSignal) => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
clearTimeout(timeoutTimer);
|
||||
if (killTimer) clearTimeout(killTimer);
|
||||
runInput.signal.removeEventListener("abort", onAbort);
|
||||
const stdout = Buffer.concat(stdoutChunks).toString("utf8");
|
||||
const stderr = Buffer.concat(stderrChunks).toString("utf8");
|
||||
const workspaceHash = workspaceIdentity(runInput.canonicalWorkspacePath);
|
||||
|
||||
if (termination === "timeout") {
|
||||
reject(new WorkspaceGitScanError(
|
||||
WORKSPACE_GIT_SCAN_ERROR_CODES.timeout,
|
||||
`Workspace Git scan timed out after ${runInput.timeoutMs}ms`,
|
||||
{ workspaceHash, timeoutMs: runInput.timeoutMs },
|
||||
));
|
||||
return;
|
||||
}
|
||||
if (termination === "cancelled") {
|
||||
reject(abortError(workspaceHash));
|
||||
return;
|
||||
}
|
||||
if (termination === "output_limit") {
|
||||
reject(new WorkspaceGitScanError(
|
||||
WORKSPACE_GIT_SCAN_ERROR_CODES.outputLimit,
|
||||
"Workspace Git scan exceeded its output limit",
|
||||
{
|
||||
workspaceHash,
|
||||
stdoutBytes,
|
||||
stderrBytes,
|
||||
maxStdoutBytes: runInput.maxStdoutBytes,
|
||||
maxStderrBytes: runInput.maxStderrBytes,
|
||||
},
|
||||
));
|
||||
return;
|
||||
}
|
||||
if (spawnError) {
|
||||
reject(new WorkspaceGitScanError(
|
||||
WORKSPACE_GIT_SCAN_ERROR_CODES.failed,
|
||||
"Workspace Git scan could not start",
|
||||
{ workspaceHash, cause: spawnError.message },
|
||||
));
|
||||
return;
|
||||
}
|
||||
if (code !== 0) {
|
||||
reject(new WorkspaceGitScanError(
|
||||
WORKSPACE_GIT_SCAN_ERROR_CODES.failed,
|
||||
"Workspace Git scan failed",
|
||||
{
|
||||
workspaceHash,
|
||||
exitCode: code,
|
||||
signal: childSignal,
|
||||
// Keep the diagnostic bounded; callers never receive raw paths in telemetry.
|
||||
stderr: stderr.trim().slice(0, 1_000),
|
||||
},
|
||||
));
|
||||
return;
|
||||
}
|
||||
resolve({ stdout, stderr });
|
||||
});
|
||||
});
|
||||
};
|
||||
}
|
||||
|
||||
export class WorkspaceGitOperationScheduler {
|
||||
@@ -478,10 +353,10 @@ export class WorkspaceGitOperationScheduler {
|
||||
if (input.signal?.aborted) throw abortError(workspaceIdentity(canonicalWorkspacePath));
|
||||
|
||||
const workspaceHash = workspaceIdentity(canonicalWorkspacePath);
|
||||
const timeoutMs = clampInteger(input.timeoutMs, this.timeoutMs, 1, 120_000);
|
||||
const timeoutMs = clampInteger(input.timeoutMs, this.timeoutMs, 1, input.onStdout ? 86_400_000 : 120_000);
|
||||
const maxStdoutBytes = clampInteger(input.maxStdoutBytes, this.maxStdoutBytes, 1, 128 * 1024 * 1024);
|
||||
const maxStderrBytes = clampInteger(input.maxStderrBytes, this.maxStderrBytes, 1, 128 * 1024 * 1024);
|
||||
const key = scanKey({
|
||||
const key = input.onStdout ? `stream:${randomUUID()}` : scanKey({
|
||||
canonicalWorkspacePath,
|
||||
args: input.args,
|
||||
env: input.env,
|
||||
@@ -489,7 +364,7 @@ export class WorkspaceGitOperationScheduler {
|
||||
maxStdoutBytes,
|
||||
maxStderrBytes,
|
||||
});
|
||||
const cacheTtlMs = clampInteger(input.cacheTtlMs, this.defaultCacheTtlMs, 0, 60_000);
|
||||
const cacheTtlMs = input.onStdout ? 0 : clampInteger(input.cacheTtlMs, this.defaultCacheTtlMs, 0, 60_000);
|
||||
// A correctness-sensitive caller that explicitly disables caching must not
|
||||
// consume a result populated earlier by the file browser.
|
||||
const cached = cacheTtlMs > 0 ? this.readCache(key) : null;
|
||||
@@ -557,6 +432,7 @@ export class WorkspaceGitOperationScheduler {
|
||||
args: [...input.args],
|
||||
fairnessKeys,
|
||||
env: input.env,
|
||||
onStdout: input.onStdout,
|
||||
timeoutMs,
|
||||
maxStdoutBytes,
|
||||
maxStderrBytes,
|
||||
@@ -590,8 +466,14 @@ export class WorkspaceGitOperationScheduler {
|
||||
};
|
||||
if (signal) {
|
||||
waiter.onAbort = () => {
|
||||
this.removeWaiter(scan, waiter);
|
||||
reject(abortError(scan.workspaceHash));
|
||||
if (scan.onStdout && scan.state === "running") {
|
||||
// The caller owns the sink storage. Do not release it until the
|
||||
// child and its last pending write settle.
|
||||
scan.controller.abort();
|
||||
} else {
|
||||
this.removeWaiter(scan, waiter);
|
||||
reject(abortError(scan.workspaceHash));
|
||||
}
|
||||
};
|
||||
signal.addEventListener("abort", waiter.onAbort, { once: true });
|
||||
}
|
||||
@@ -678,6 +560,7 @@ export class WorkspaceGitOperationScheduler {
|
||||
canonicalWorkspacePath: scan.canonicalWorkspacePath,
|
||||
args: scan.args,
|
||||
env: scan.env,
|
||||
onStdout: scan.onStdout,
|
||||
signal: scan.controller.signal,
|
||||
timeoutMs: scan.timeoutMs,
|
||||
killGraceMs: this.killGraceMs,
|
||||
@@ -887,6 +770,8 @@ setExpensiveWorkspaceGitExecutor(async (input) => {
|
||||
// referenced-project scan sets this to its hardened environment, so the
|
||||
// hardening survives the hop through this shared scheduler.
|
||||
env: input.env,
|
||||
onStdout: input.onStdout,
|
||||
signal: input.signal,
|
||||
});
|
||||
return { stdout: result.stdout, stderr: result.stderr };
|
||||
});
|
||||
|
||||
@@ -0,0 +1,142 @@
|
||||
import { execFile } from "node:child_process";
|
||||
import fs from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { promisify } from "node:util";
|
||||
import { afterEach, expect, it } from "vitest";
|
||||
import { disposeGitWorkspaceSnapshot, readGitWorkspaceSnapshot, runLocalGit, setExpensiveWorkspaceGitExecutor, type GitWorkspaceSnapshot } from "@paperclipai/adapter-utils/git-workspace-sync";
|
||||
import { isPathManifest, workspacePaths } from "@paperclipai/adapter-utils/workspace-manifest";
|
||||
import { prepareSandboxManagedRuntime, type PreparedSandboxManagedRuntime, type SandboxManagedRuntimeClient } from "@paperclipai/adapter-utils/sandbox-managed-runtime";
|
||||
import { createWorkspaceGitOperationScheduler } from "./workspace-git-operation-scheduler.js";
|
||||
|
||||
const execute = promisify(execFile);
|
||||
const directories: string[] = [];
|
||||
const snapshots: GitWorkspaceSnapshot[] = [];
|
||||
const runtimes: PreparedSandboxManagedRuntime[] = [];
|
||||
afterEach(async () => {
|
||||
setExpensiveWorkspaceGitExecutor(null);
|
||||
for (const runtime of runtimes.splice(0)) await runtime.cleanupWorkspaceSnapshot();
|
||||
for (const snapshot of snapshots.splice(0)) await disposeGitWorkspaceSnapshot(snapshot);
|
||||
for (const directory of directories.splice(0)) await fs.rm(directory, { recursive: true, force: true });
|
||||
}, 60_000);
|
||||
|
||||
function localClient(commands: string[]): SandboxManagedRuntimeClient {
|
||||
return {
|
||||
makeDir: async (dir) => { await fs.mkdir(dir, { recursive: true }); },
|
||||
writeFile: async (file, bytes) => { await fs.writeFile(file, Buffer.from(bytes)); },
|
||||
readFile: async (file) => fs.readFile(file),
|
||||
remove: async (file) => { await fs.rm(file, { recursive: true, force: true }); },
|
||||
run: async (command) => { await execute("sh", ["-c", command]); },
|
||||
syncIn: async (operations) => {
|
||||
for (const operation of operations) {
|
||||
for (const mapping of operation.files) {
|
||||
await fs.mkdir(path.dirname(mapping.targetPath), { recursive: true });
|
||||
await fs.copyFile(mapping.sourcePath, mapping.targetPath);
|
||||
}
|
||||
for (const command of operation.postUploadCommands ?? []) {
|
||||
commands.push(command.command);
|
||||
await execute("sh", ["-c", command.command], { maxBuffer: 64 * 1024 });
|
||||
}
|
||||
}
|
||||
return { operations: [] };
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
it("streams and stages all four real Git filename lanes above 32 MiB through the shared scheduler", async () => {
|
||||
const root = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-streaming-real-git-"));
|
||||
directories.push(root);
|
||||
const repo = path.join(root, "repo");
|
||||
await fs.mkdir(repo);
|
||||
const env = { ...process.env, GIT_AUTHOR_NAME: "Test", GIT_COMMITTER_NAME: "Test", GIT_AUTHOR_EMAIL: "test@example.test", GIT_COMMITTER_EMAIL: "test@example.test" };
|
||||
const git = (args: string[]) => runLocalGit(repo, args, { env, timeout: 120_000, maxBuffer: 64 * 1024 });
|
||||
await git(["init"]);
|
||||
await git(["commit", "--allow-empty", "-qm", "fixture"]);
|
||||
const parent = path.join(repo, ...Array.from({ length: 4 }, () => "nested-".repeat(30)), "storybook-output");
|
||||
await fs.mkdir(parent, { recursive: true });
|
||||
const fileName = (index: number) => `${"asset-".repeat(36)}${index}.js`;
|
||||
const writeFiles = async (content: string) => {
|
||||
for (let start = 0; start < 40_000; start += 100) {
|
||||
await Promise.all(Array.from({ length: 100 }, (_, offset) => fs.writeFile(path.join(parent, fileName(start + offset)), content)));
|
||||
}
|
||||
};
|
||||
await writeFiles("base");
|
||||
const unusual = [" space ", "line\nbreak", "-option", "雪-💾", "wild[?]*"];
|
||||
for (const name of unusual) await fs.writeFile(path.join(repo, name), name);
|
||||
const scheduler = createWorkspaceGitOperationScheduler({ concurrency: 2, queueCapacity: 8 });
|
||||
const bindScheduler = () => setExpensiveWorkspaceGitExecutor((input) => scheduler.run({
|
||||
workspacePath: input.localDir, args: input.args, operation: input.operation,
|
||||
timeoutMs: input.timeout, onStdout: input.onStdout, signal: input.signal, cacheTtlMs: 0,
|
||||
maxStdoutBytes: 1, maxStderrBytes: 64 * 1024,
|
||||
}));
|
||||
const scan = async () => {
|
||||
const result = await readGitWorkspaceSnapshot(repo);
|
||||
expect(result).not.toBeNull();
|
||||
snapshots.push(result!);
|
||||
expect(JSON.stringify(result).length).toBeLessThan(4096);
|
||||
return result!;
|
||||
};
|
||||
const checkLargeLane = (paths: GitWorkspaceSnapshot["overlayPaths"]) => {
|
||||
let count = 0, bytes = 0;
|
||||
for (const relative of workspacePaths(paths)) {
|
||||
if (!relative.endsWith(".js")) continue;
|
||||
count++;
|
||||
bytes += Buffer.byteLength(relative) + 1;
|
||||
}
|
||||
expect(count).toBe(40_000);
|
||||
expect(bytes).toBe(43_428_890);
|
||||
};
|
||||
// Standalone callers must use the same streaming semantics.
|
||||
setExpensiveWorkspaceGitExecutor(null);
|
||||
const standalone = await scan();
|
||||
checkLargeLane(standalone.overlayPaths);
|
||||
await disposeGitWorkspaceSnapshot(standalone);
|
||||
bindScheduler();
|
||||
const snapshot = await scan();
|
||||
checkLargeLane(snapshot.overlayPaths);
|
||||
await fs.writeFile(path.join(parent, "late.js"), "must not stage");
|
||||
const commands: string[] = [];
|
||||
const remote = path.join(root, "remote");
|
||||
const stage = async (gitSnapshot: GitWorkspaceSnapshot) => {
|
||||
const runtime = await prepareSandboxManagedRuntime({
|
||||
spec: { transport: "sandbox", provider: "test", sandboxId: "test", remoteCwd: remote, timeoutMs: 120_000, apiKey: null },
|
||||
client: localClient(commands), adapterKey: "stream-test", workspaceLocalDir: repo, workspaceGitSnapshot: gitSnapshot,
|
||||
});
|
||||
runtimes.push(runtime);
|
||||
return runtime;
|
||||
};
|
||||
const runtime = await stage(snapshot);
|
||||
const remoteParent = path.join(remote, path.relative(repo, parent));
|
||||
let stagedCount = 0;
|
||||
for await (const entry of await fs.opendir(remoteParent)) {
|
||||
expect(await fs.readFile(path.join(remoteParent, entry.name), "utf8")).toBe("base");
|
||||
stagedCount++;
|
||||
}
|
||||
expect(stagedCount).toBe(40_000);
|
||||
for (const name of unusual) expect(await fs.readFile(path.join(remote, name), "utf8")).toBe(name);
|
||||
const manifestFile = isPathManifest(snapshot.overlayPaths) ? snapshot.overlayPaths.filePath : "";
|
||||
await runtime.cleanupWorkspaceSnapshot();
|
||||
await expect(fs.stat(manifestFile)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
await fs.rm(path.join(parent, "late.js"));
|
||||
|
||||
await fs.writeFile(path.join(repo, ".gitignore"), "*.js\n");
|
||||
const ignored = await scan();
|
||||
checkLargeLane(ignored.ignoredPaths);
|
||||
await disposeGitWorkspaceSnapshot(ignored);
|
||||
await fs.rm(path.join(repo, ".gitignore"));
|
||||
await git(["add", "--", "."]);
|
||||
await git(["commit", "-qm", "track generated files"]);
|
||||
await writeFiles("changed");
|
||||
const changed = await scan();
|
||||
checkLargeLane(changed.overlayPaths);
|
||||
await disposeGitWorkspaceSnapshot(changed);
|
||||
await fs.rm(parent, { recursive: true });
|
||||
const deleted = await scan();
|
||||
checkLargeLane(deleted.deletedPaths);
|
||||
await stage(deleted);
|
||||
let remaining = 0;
|
||||
for await (const _entry of await fs.opendir(remoteParent)) remaining++;
|
||||
expect(remaining).toBe(0);
|
||||
expect(commands.every((command) => command.length < 16 * 1024)).toBe(true);
|
||||
expect(scheduler.snapshot()).toMatchObject({ activeCount: 0, queuedCount: 0, inFlightCount: 0, cacheBytes: 0 });
|
||||
}, 180_000);
|
||||
Reference in new issue
Block a user