RomanNum3ral
/
paperclip
mirror of https://github.com/paperclipai/paperclip.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat: parallelize sandbox file-sync behind a provider opt-in capability (#11736) 

## Thinking Path

> - Paperclip runs AI agents through local and remote execution
adapters.
> - Sandbox providers move workspace and asset files before and after
agent runs.
> - Serial file transfers delay startup and teardown when several
operations do not depend on each other.
> - Providers need an opt-in contract so existing providers keep their
serial behavior.
> - This pull request adds a bounded scheduler and routes inbound and
outbound sync operations through it.
> - The benefit is shorter sandbox setup and teardown with stable
errors, clear telemetry, and a safe opt-in path.

## Linked Issues or Issue Description

**Subsystem affected**

Cross-cutting (multiple of the above): packages/shared,
packages/adapter-utils, packages/plugins, and server.

**Problem or motivation**

Sandbox sync processes the workspace, assets, and referenced projects in
series. This adds avoidable wait time to agent startup and teardown.

**Proposed solution**

Add a fail-closed provider capability named concurrentSyncOperations.
Use a bounded scheduler with a limit of four operations. Preserve
operation order for error reporting. Keep non-opted-in providers on the
serial path.

**Alternatives considered**

Increase the serial transfer speed or add provider-specific schedulers.
Those options do not provide one shared contract or stable behavior
across providers.

**Roadmap alignment**

ROADMAP.md lists cloud and sandbox agents as a product area. This change
improves sandbox execution without changing the control-plane contract.

**Additional context**

The Daytona provider opts in. Board trials on this commit showed overlap
for inbound sync and outbound restore, with no referenced-project
staging failures.

## What Changed

- Add the concurrentSyncOperations sandbox capability and fail-closed
parsing.
- Add a bounded settle-all scheduler with stable input-order errors.
- Parallelize inbound workspace, asset, and referenced-project sync
operations when the provider opts in.
- Parallelize outbound workspace and asset restore operations when the
provider opts in.
- Surface referenced-project failure text in run logs and server
telemetry.
- Add Daytona sync spans and the capability declaration.
- Preserve in-flight upload scratch tarballs during workspace wipe.
- Add unit and regression tests for the scheduler, coordinators,
provider behavior, telemetry, and wipe race.

## Verification

- Run the adapter-utils and server type checks.
- Run the targeted adapter-utils, server, and Daytona test suites.
- Run the full automated sweep.
- Review six cold Daytona trials, with three serial and three parallel
runs.
- Confirm that parallel trials show inbound overlap and outbound restore
overlap.
- Confirm that providers without the capability keep serial behavior.

## Risks

- Providers must opt in only when their file operations can run safely
at the same time.
- A provider that declares the capability incorrectly can expose
transfer races.
- The scheduler keeps a limit of four to bound resource use.
- Providers without the capability keep the prior serial behavior.

## Model Used

OpenAI GPT-5 in the Codex runtime. The model used tool calls, code
inspection, and GitHub workflow support. The model did not author the
implementation commits.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with Fixes: # / Closes #
/ Refs # OR (b) described the issue in-PR following the relevant issue
template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub #NNN / github.com/paperclipai/paperclip URLs)
- [x] My branch name describes the change and contains no internal
Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Nicky Leach 2026-08-19 12:34:11 -07:00 committed by GitHub
parent a5f3ac6b48
commit 233be4b36c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
30 changed files with 3032 additions and 478 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

157
packages/adapter-utils/src/acpx-engine/execute.test.ts
View File

@ -2594,6 +2594,42 @@ describe("ACPX engine remote managed-home seam (PR 2: per-adapter home seed)", (
expect(stageArgs.assets ?? []).toEqual([]);
expect(sessionInputs[0]?.cwd).toBe(remoteCwd);
});
it("test_remote_seam_surfaces_referenced_project_staging_failure_reason", async () => {
// A referenced project that failed to stage into the sandbox is a first-class
// run outcome. The run result carries the failure reason for each dropped
// project, and the run log gains one stderr line per failure that names the
// project id and the reason. The run stays successful (per-project isolation).
const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox();
const { result, logs } = await runExecutor(
{ agent: "custom", agentCommand: "node ./fake-acp.js", stateDir, cwd: localCwd },
{
authToken: "real-run-jwt",
executionTarget,
prepareRemoteManagedHome: async (input) => {
const stagedRuntime = await input.stage([]);
// Inject a per-project staging failure the coordinator would record on a
// real sandbox extract failure.
stagedRuntime.additionalSourceFailures = [
{ projectId: "proj-x", error: "extract failed: boom" },
];
return { stagedRuntime };
},
},
);
// The run result carries the failure with its reason.
expect(result.referencedProjectStagingFailures).toEqual([
{ projectId: "proj-x", error: "extract failed: boom" },
]);
// The run log gains one stderr line that names the project id and the reason.
const failureLine = logs.find(
(entry) => entry.stream === "stderr" && entry.text.includes("proj-x"),
);
expect(failureLine?.text).toBe(
"[paperclip] Referenced project proj-x failed to stage; the run continues without it: extract failed: boom\n",
);
});
});
describe("ACPX engine remote session-lifecycle re-staging (PR 3: stage once / reuse on compatible resume)", () => {
@ -3349,14 +3385,27 @@ describe("ACPX engine sandbox-start spans (opt-in root + child parenting)", () =
expect(turnSpan!.parent).toBe(runRootSpan);
expect(turnSpan!.ended).toBe(true);
// The settlement sync-back opens the `sandbox.syncBack` span. It runs at
// teardown, so it parents to the run root span, not to the bring-up span.
const syncBackSpan = spans.find((span) => span.name === "sandbox.syncBack");
expect(syncBackSpan).toBeTruthy();
expect(syncBackSpan!.parent).toBe(runRootSpan);
expect(syncBackSpan!.ended).toBe(true);
// A codex bring-up over the remote sandbox lane crosses all 7 boundaries.
// Each boundary span parents to the sandbox bring-up span, not to the run
// root or the turn span. The `stage.sync` step also opens three host
// sub-step spans — `snapshot.git`, `snapshot.baseline`, and `pack` — around
// its git enumeration, baseline content-hash walk, and workspace tarball
// build, so those nest one level deeper.
// root or the turn span. The `stage.sync` step also opens the two pre-task
// sub-step spans — `snapshot.git` and `snapshot.baseline` — plus the
// `stage.workspace` inbound task span, and the `pack` span nests one level
// deeper under `stage.workspace`.
const childNames = spans
.filter((span) => span !== runRootSpan && span !== startupSpan && span !== turnSpan)
.filter(
(span) =>
span !== runRootSpan &&
span !== startupSpan &&
span !== turnSpan &&
span !== syncBackSpan,
)
.map((span) => span.name)
.sort();
expect(childNames).toEqual(
@ -3370,32 +3419,44 @@ describe("ACPX engine sandbox-start spans (opt-in root + child parenting)", () =
"snapshot.baseline",
"snapshot.git",
"stage.sync",
"stage.workspace",
"workspace.resolve",
],
);
// The three host sub-step spans nest under the `stage.sync` step span (that
// host work runs inside the step), not directly under the bring-up span.
// The two pre-task sub-step spans and the `stage.workspace` task span nest
// under the `stage.sync` step span (that host work runs inside the step), not
// directly under the bring-up span. The `pack` span nests under
// `stage.workspace`, because the host builds the tarball inside the task.
const stageSyncSpan = spans.find((span) => span.name === "stage.sync");
const stageWorkspaceSpan = spans.find((span) => span.name === "stage.workspace");
const packSpan = spans.find((span) => span.name === "pack");
const snapshotGitSpan = spans.find((span) => span.name === "snapshot.git");
const snapshotBaselineSpan = spans.find((span) => span.name === "snapshot.baseline");
expect(stageSyncSpan).toBeTruthy();
expect(stageWorkspaceSpan).toBeTruthy();
expect(packSpan).toBeTruthy();
expect(snapshotGitSpan).toBeTruthy();
expect(snapshotBaselineSpan).toBeTruthy();
expect(packSpan!.parent).toBe(stageSyncSpan);
expect(stageWorkspaceSpan!.parent).toBe(stageSyncSpan);
expect(packSpan!.parent).toBe(stageWorkspaceSpan);
expect(snapshotGitSpan!.parent).toBe(stageSyncSpan);
expect(snapshotBaselineSpan!.parent).toBe(stageSyncSpan);
expect(packSpan!.ended).toBe(true);
// Every boundary step span parents to the sandbox bring-up span and ends.
// The three `stage.sync` sub-step spans are the exceptions: they parent to
// `stage.sync` above.
const stageSyncChildren = new Set([packSpan, snapshotGitSpan, snapshotBaselineSpan]);
// The `stage.sync` sub-step spans and the run-parented `sandbox.syncBack`
// span are the exceptions: they parent above.
const nonStartupChildren = new Set([
packSpan,
stageWorkspaceSpan,
snapshotGitSpan,
snapshotBaselineSpan,
syncBackSpan,
]);
for (const span of spans) {
if (span === runRootSpan || span === startupSpan || span === turnSpan) continue;
if (stageSyncChildren.has(span)) continue;
if (nonStartupChildren.has(span)) continue;
expect(span.parent, `span "${span.name}" must parent to the startup span`).toBe(startupSpan);
expect(span.ended, `span "${span.name}" must end`).toBe(true);
}
@ -4139,6 +4200,78 @@ describe("ACPX engine sandbox-start spans (opt-in root + child parenting)", () =
expect(execParents.size).toBe(4);
});
it("test_sync_back_runs_inside_sandbox_syncback_span", async () => {
const root = await makeTempRoot();
const stateDir = path.join(root, "state");
const localCwd = path.join(root, "worktree");
const codexHome = path.join(root, "codex-home");
await fs.mkdir(localCwd, { recursive: true });
await fs.mkdir(codexHome, { recursive: true });
const executionTarget = await remoteSandboxTarget(root);
const { traceContext, spans } = createRecordingStartupTrace();
// The settlement `syncBack` step runs the managed-home teardown. The teardown
// issues one exec through the same host-to-sandbox seam the run uses. The
// recorder captures the exec parent, so the test proves the teardown runs
// inside the run-parented `sandbox.syncBack` span.
let teardownExecFired = false;
const execute = createAcpxEngineExecutor({
prepareRemoteManagedHome: async (input) => {
const stagedRuntime = await input.stage([]);
return {
stagedRuntime,
teardown: async () => {
if (!teardownExecFired) {
teardownExecFired = true;
issueSandboxExecFromStore(traceContext);
}
},
};
},
createRuntime: () => buildRuntime() as never,
});
const result = await execute({
runId: "run-sync-back-span",
agent: { id: "agent-1", companyId: "company-1" },
runtime: {},
config: {
agent: "codex",
agentCommand: "node ./fake-acp.js",
stateDir,
cwd: localCwd,
env: { CODEX_HOME: codexHome },
},
context: {},
authToken: "real-run-jwt",
executionTarget,
startupTraceContext: traceContext,
onLog: async () => {},
onMeta: async () => {},
onEvent: async () => {},
} as never);
expect(result.exitCode).toBe(0);
expect(teardownExecFired).toBe(true);
// One `sandbox.syncBack` span opens, ends, and parents to `task.run`. The
// settlement runs the teardown after the turn, so the run parent is
// `task.run` at that time.
const runSpan = spans.find((span) => span.name === "task.run");
expect(runSpan).toBeTruthy();
const syncBackSpans = spans.filter((span) => span.name === "sandbox.syncBack");
expect(syncBackSpans).toHaveLength(1);
const syncBackSpan = syncBackSpans[0]!;
expect(syncBackSpan.ended).toBe(true);
expect(syncBackSpan.parent).toBe(runSpan);
// The teardown exec parents to `sandbox.syncBack`, so the sync-back copy-back
// runs inside the span, not unparented at teardown.
const syncBackExec = spans.find(
(span) => span.name === "sandbox.exec" && span.parent === syncBackSpan,
);
expect(syncBackExec, "the teardown exec must parent to sandbox.syncBack").toBeTruthy();
});
it("test_no_sandbox_exec_parents_to_http_root", async () => {
const root = await makeTempRoot();
const stateDir = path.join(root, "state");

24
packages/adapter-utils/src/acpx-engine/execute.ts
View File

@ -3326,7 +3326,7 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
let resumedSession = false;
let clearSession = false;
let referencedProjectStagingFailuresField:
| { referencedProjectStagingFailures: Array<{ projectId: string }> }
| { referencedProjectStagingFailures: Array<{ projectId: string; error: string }> }
| Record<string, never> = {};
const recordTeardownError = async (step: string, teardownErr: unknown) => {
const reason = teardownErr instanceof Error ? teardownErr.message : String(teardownErr);
@ -3432,9 +3432,19 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
// is a failure to report.
const referencedProjectStagingFailures = (
prepared.stagedRuntime?.additionalSourceFailures ?? []
).map((failure) => ({ projectId: failure.projectId }));
).map((failure) => ({ projectId: failure.projectId, error: failure.error }));
referencedProjectStagingFailuresField =
referencedProjectStagingFailures.length > 0 ? { referencedProjectStagingFailures } : {};
// Write one run-log line per failure, so a reader of the run log alone sees
// each dropped referenced project and its reason. The run continues without
// the failed project (per-project failure isolation). Goes to stderr: the
// acpx stdout log stream carries machine-parseable JSON event payloads.
for (const failure of referencedProjectStagingFailures) {
await ctx.onLog(
"stderr",
`[paperclip] Referenced project ${failure.projectId} failed to stage; the run continues without it: ${failure.error}\n`,
);
}
// State the effective wall-clock timeout and its source up front so a
// later timeout is diagnosable from the run log alone. Goes to stderr:
// the acpx stdout log stream carries JSON acpx.* event payloads and must
@ -4226,9 +4236,15 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
stopTransport: () => timedPhase("stop_transport", async () => {
await stopRunTransport(prepared);
}),
// The site sync-back (the managed-home copy-back).
// The site sync-back (the managed-home copy-back). The run-parented span
// runner wraps the restore in a `sandbox.syncBack` span. The runner also
// publishes the run parent into the runtime-parent store while the restore
// runs, so the host mints a `traceparent` for the provider spans, and the
// per-task restore spans parent to `sandbox.syncBack`.
syncBack: () => timedPhase("sync_back", async () => {
await syncBackManagedHome(prepared);
await runRuntimeSpan("sandbox.syncBack", async () => {
await syncBackManagedHome(prepared);
});
}),
// The staging lease releases as the run's final act, AFTER the coordinator
// reproduces the result, in the run root `finally` below. This step stays a

2
packages/adapter-utils/src/acpx-engine/run-contracts.ts
View File

@ -325,6 +325,8 @@ export interface SitePlan {
/** One referenced project whose staging into the sandbox failed. */
export interface ReferencedProjectStagingFailure {
readonly projectId: string;
/** The failure reason. A reader of the run sees why the project dropped. */
readonly error: string;
}
/**

57
packages/adapter-utils/src/acpx-engine/run-site-sandbox.test.ts
View File

@ -140,6 +140,23 @@ describe("sandbox run site", () => {
expect(scopeOf("agent_bridge")).toBe("per_run");
});
it("test_sandbox_site_place_workspace_carries_referenced_project_failure_reason", async () => {
// A referenced project that fails to stage carries its failure reason back on
// the placed-workspace result, so a reader of the run learns why it dropped.
const failedStaged = {
runtimeRootDir: "/remote/fail/.paperclip-runtime/acpx",
additionalSourceDirs: {},
additionalSourceFailures: [{ projectId: "proj-x", error: "extract failed: boom" }],
} as unknown as PreparedAdapterExecutionTargetRuntime;
const { site } = makeSite({ stage: async () => failedStaged });
const placed = await site.placeWorkspace(makeContext("s"));
expect(placed).toEqual({
referencedProjectStagingFailures: [{ projectId: "proj-x", error: "extract failed: boom" }],
});
});
it("test_sandbox_site_preserves_bridge_overlap_and_callback_sequencing", async () => {
const events: string[] = [];
let releasePaperclip!: () => void;
@ -181,6 +198,46 @@ describe("sandbox run site", () => {
expect(transport.launchEnv).toEqual(processLaunchEnv);
});
it("test_sandbox_site_session_new_starts_only_after_the_sync_barrier_settles", async () => {
// The staging step wraps the inbound sync coordinator. Hold it open with a
// gate, so the sync barrier stays unsettled. `placeWorkspace` awaits staging,
// so it stays pending while the gate is held.
let releaseStaging!: () => void;
const stagingGate = new Promise<void>((resolve) => {
releaseStaging = resolve;
});
const freshStaged = makeStagedRuntime("fresh");
const { site, bridgeCalls } = makeSite({
stage: async () => {
await stagingGate;
return freshStaged;
},
});
// Drive the real run order: the engine awaits `placeWorkspace`, then starts the
// transport. `startProcessSessionBridge` models the ACP `session/new` startup.
let placed = false;
const run = (async () => {
await site.placeWorkspace(makeContext("s"));
placed = true;
await site.startTransport(makeContext("s"));
})();
run.catch(() => undefined);
// The sync barrier is still open, so `placeWorkspace` has not resolved and the
// process-session bridge that starts `session/new` has not started.
await new Promise<void>((resolve) => setTimeout(resolve, 50));
expect(placed).toBe(false);
expect(bridgeCalls).not.toContain("process-session:start");
// Settle the sync barrier. `placeWorkspace` resolves, then the transport starts
// `session/new`.
releaseStaging();
await run;
expect(placed).toBe(true);
expect(bridgeCalls).toContain("process-session:start");
});
it("test_sandbox_site_borrow_yields_files_and_runtime_is_still_created", async () => {
const cachedRuntime = makeStagedRuntime("cached");
const stagedRuntimes = new Map<string, StagedRuntimeStoreEntry>([

2
packages/adapter-utils/src/acpx-engine/run-site-sandbox.ts
View File

@ -320,7 +320,7 @@ export function createSandboxRunSite(options: SandboxRunSiteOptions): SandboxRun
}
return {
referencedProjectStagingFailures: (staged.stagedRuntime.additionalSourceFailures ?? []).map(
(failure) => ({ projectId: failure.projectId }),
(failure) => ({ projectId: failure.projectId, error: failure.error }),
),
};
},

5
packages/adapter-utils/src/acpx-engine/startup-timing.ts
View File

@ -109,6 +109,11 @@ export const SANDBOX_STARTUP_SPAN_ATTRS = {
transferWallMs: `${SANDBOX_STARTUP_SPAN_ATTR_PREFIX}transfer.wall_ms`,
/** The number of serial guard round trips before one transfer. */
transferGuardCount: `${SANDBOX_STARTUP_SPAN_ATTR_PREFIX}transfer.guard.count`,
/** The transfer direction: `inbound` for an upload to the sandbox, `outbound`
* for a download from the sandbox. The parent span carries operation identity,
* so the transfer span never carries an operation label. The value stays in a
* closed set, so the attribute cardinality is bounded. */
transferDirection: `${SANDBOX_STARTUP_SPAN_ATTR_PREFIX}transfer.direction`,
} as const;
/** The closed value set for the `outcome` attribute. */

56
packages/adapter-utils/src/command-managed-runtime.test.ts
View File

@ -552,6 +552,62 @@ describe("command managed runtime", () => {
expect(native.syncOut).toBeTypeOf("function");
});
it("base64 fallback client reports allowConcurrentSyncOperations true", () => {
// A runner with no native sync uses the base64 fallback, which always
// permits concurrent sync operations.
const base = makeSpawnRunner().runner;
const client = createCommandManagedRuntimeClient({ runner: base, commandCwd: "/", timeoutMs: 1 });
expect(client.allowConcurrentSyncOperations).toBe(true);
});
it("undeclared native runner reports allowConcurrentSyncOperations false", () => {
// A native runner (both sync verbs) that never opted into concurrency keeps
// the flag off.
const base = makeSpawnRunner().runner;
const undeclaredNative: CommandManagedRuntimeRunner = {
...base,
syncIn: async () => ({ operations: [] }),
syncOut: async () => ({ operations: [] }),
};
const client = createCommandManagedRuntimeClient({
runner: undeclaredNative,
commandCwd: "/",
timeoutMs: 1,
});
expect(client.allowConcurrentSyncOperations).toBe(false);
});
it("declared native runner reports allowConcurrentSyncOperations true", () => {
// A native runner that verified the opt-in carries the flag through to the
// client.
const base = makeSpawnRunner().runner;
const declaredNative: CommandManagedRuntimeRunner = {
...base,
allowConcurrentSyncOperations: true,
syncIn: async () => ({ operations: [] }),
syncOut: async () => ({ operations: [] }),
};
const client = createCommandManagedRuntimeClient({
runner: declaredNative,
commandCwd: "/",
timeoutMs: 1,
});
expect(client.allowConcurrentSyncOperations).toBe(true);
});
it("base64 fallback ignores a runner opt-in without both sync verbs", () => {
// A runner that sets the opt-in but exposes only one sync verb still uses
// the fallback, which permits concurrency independent of the runner flag.
const base = makeSpawnRunner().runner;
const onlyIn: CommandManagedRuntimeRunner = {
...base,
allowConcurrentSyncOperations: true,
syncIn: async () => ({ operations: [] }),
};
const client = createCommandManagedRuntimeClient({ runner: onlyIn, commandCwd: "/", timeoutMs: 1 });
expect(client.allowConcurrentSyncOperations).toBe(true);
});
it("test_client_syncIn_delegates_to_native_runner_with_zero_execute_calls", async () => {
// With a native runner, `client.syncIn` forwards `files` + `postUploadCommands`
// to the runner and issues ZERO `execute` round-trips (the provider owns the

16
packages/adapter-utils/src/command-managed-runtime.ts
View File

@ -19,6 +19,15 @@ import type { RuntimeProgressSink, RuntimeStatusSink } from "./runtime-progress.
import type { RuntimeSpanRunner } from "./acpx-engine/startup-timing.js";
export interface CommandManagedRuntimeRunner {
/**
* True when the provider verified the concurrent-sync opt-in. A native runner
* carries the value from the effective capability snapshot
* (`concurrentSyncOperations`). The client copies it onto the prepared sync
* client only on the native path; the base64 fallback ignores it and always
* permits concurrency. The default is false, so an undeclared native provider
* never permits concurrent sync operations.
*/
allowConcurrentSyncOperations?: boolean;
/**
* True only when `execute({ stdin })` can surface useful in-flight progress
* for a single stdin-backed command. Provider-backed sandbox runners usually
@ -446,6 +455,13 @@ export function createCommandManagedRuntimeClient(input: {
const nativeSyncIn = input.runner.syncIn;
const nativeSyncOut = input.runner.syncOut;
const hasNativeBoth = Boolean(nativeSyncIn && nativeSyncOut);
// The base64 fallback always permits concurrent sync operations. A native
// runner permits them only when the provider verified the opt-in; an
// undeclared native provider keeps concurrency off. One flag serves both sync
// directions.
client.allowConcurrentSyncOperations = hasNativeBoth
? input.runner.allowConcurrentSyncOperations === true
: true;
client.syncIn = async (operations) => {
assertPostUploadCommandsConfined(operations);
if (hasNativeBoth) {

2
packages/adapter-utils/src/execution-target.ts
View File

@ -115,6 +115,7 @@ export interface EffectiveSandboxCapabilities {
readonly persistentProcessSessions: boolean;
readonly independentControlCommands: boolean;
readonly incrementalSessionOutput: boolean;
readonly concurrentSyncOperations: boolean;
}
export interface AdapterSandboxExecutionTarget extends AdapterExecutionTargetWorkspaceMetadata {
@ -254,6 +255,7 @@ function parseEffectiveSandboxCapabilities(value: unknown): EffectiveSandboxCapa
persistentProcessSessions: parsed.persistentProcessSessions === true,
independentControlCommands: parsed.independentControlCommands === true,
incrementalSessionOutput: parsed.incrementalSessionOutput === true,
concurrentSyncOperations: parsed.concurrentSyncOperations === true,
};
}

1078
packages/adapter-utils/src/sandbox-managed-runtime.test.ts
View File

File diff suppressed because it is too large Load Diff

911
packages/adapter-utils/src/sandbox-managed-runtime.ts
View File

@ -27,6 +27,11 @@ import {
type RuntimeStatusSink,
} from "./runtime-progress.js";
import { isRelativePathOrDescendant, shouldExcludePath } from "./exclude-patterns.js";
import {
scheduleSyncOperations,
SYNC_OPERATION_CONCURRENCY_LIMIT,
type SyncOperationTask,
} from "./sync-operation-schedule.js";
import type { RuntimeSpanRunner } from "./acpx-engine/startup-timing.js";
const execFile = promisify(execFileCallback);
@ -102,11 +107,18 @@ export interface SandboxManagedRuntimeAssetProvision {
/**
* Context passed to an asset's `restore` contribution during teardown.
* `assetDir` is the asset's directory inside the sandbox and `readFile` reads
* a file back from the sandbox as raw bytes.
* a file back from the sandbox as raw bytes. `tempDir` is a host scratch
* directory that belongs to this restore task alone. The shared scheduler can
* run restore tasks at the same time, so a task must not share scratch space
* with another task. A restore that needs a host temporary file writes it under
* `tempDir`. The coordinator removes the directory after the task settles. The
* sandbox coordinator always sets `tempDir`. A serial runtime that never runs
* restore tasks at the same time can omit it.
*/
export interface SandboxManagedRuntimeAssetRestoreContext {
assetDir: string;
readFile: (remotePath: string) => Promise<Buffer>;
tempDir?: string;
}
export interface SandboxManagedRuntimeAsset {
@ -244,6 +256,15 @@ export interface SandboxManagedRuntimeClient {
listFiles(remotePath: string): Promise<string[]>;
remove(remotePath: string): Promise<void>;
run(command: string, options: { timeoutMs: number }): Promise<void>;
/**
* True when the orchestrator may run this client's sync operations
* concurrently. The base64 fallback always sets it true. A native provider
* takes the value from the verified `concurrentSyncOperations` opt-in; an
* undeclared native provider keeps it false. One flag serves both `syncIn` and
* `syncOut`. `createCommandManagedRuntimeClient` always sets it on a prepared
* client; it is optional here so a test mock can omit it.
*/
allowConcurrentSyncOperations?: boolean;
/**
* Optional native inbound transfer. Present only when the sandbox provider
* advertises both `environmentSyncIn` and `environmentSyncOut`; otherwise the
@ -358,9 +379,18 @@ function buildWorkspaceTarExtractCommand(input: {
remoteTar: string;
wipeExceptNames: string[] | null;
}): string {
// The wipe must also preserve any in-flight sync scratch tarball at the
// workspace root. A concurrent referenced-project upload stages a scratch
// tarball named `.paperclip-upload-<uuid>.tar` there. Without this preserve
// term the wipe unlinks the in-flight tarball and the later extract fails.
// The static pattern must agree with the daytona scratch prefix
// `SCRATCH_PREFIX` in
// `packages/plugins/sandbox-providers/daytona/src/file-sync.ts:80`.
// The term is a static literal; `preserveFindArgs` shell-quotes it, so the
// shell passes it to `find -name` as a pattern (Security Conditions C1/C3).
const wipe = input.wipeExceptNames
? ` && find ${shellQuote(input.workspaceRemoteDir)} -mindepth 1 -maxdepth 1 ` +
`${preserveFindArgs(input.wipeExceptNames)} -exec rm -rf -- {} +`
`${preserveFindArgs([...input.wipeExceptNames, ".paperclip-upload-*"])} -exec rm -rf -- {} +`
: "";
return (
`mkdir -p ${shellQuote(input.workspaceRemoteDir)}${wipe} && ` +
@ -384,6 +414,29 @@ function buildUniqueStagingPath(input: { targetPath: string; suffix: string }):
return `${input.targetPath}${input.suffix}.${randomUUID()}`;
}
// The workspace stages under `<runtimeRootDir>/workspace-upload.tar` and, for a
// git-backed workspace, under `<runtimeRootDir>/git-workspace-upload.tar`. Each
// asset stages under `<runtimeRootDir>/<key>-upload.tar`, so an asset key equal to
// one of these stems resolves to the same remote archive path. Reserve the stems.
const RESERVED_RUNTIME_ASSET_KEYS = new Set(["workspace", "git-workspace"]);
// Reject an asset key before any path is built from it. An asset key becomes a
// remote directory (`<runtimeRootDir>/<key>`), a remote archive name
// (`<key>-upload.tar`), and a host temp file (`<key>.tar`). A path separator or
// `..` in the key escapes those roots. A reserved stem makes the asset archive
// share a path with the workspace archive; under concurrent sync the asset task
// and the workspace task then write or upload the same archive at the same time,
// which fails extraction nondeterministically or puts asset bytes in the
// workspace. Fail closed on both cases.
function assertRuntimeAssetKeyIsSafe(key: string): void {
if (key.length === 0 || key.includes("/") || key.includes("\\") || key.includes("..")) {
throw new Error(`sandbox runtime asset key is not a simple path segment: ${key}`);
}
if (RESERVED_RUNTIME_ASSET_KEYS.has(key)) {
throw new Error(`sandbox runtime asset key collides with a reserved runtime archive name: ${key}`);
}
}
export function parseSandboxRemoteExecutionSpec(value: unknown): SandboxRemoteExecutionSpec | null {
const parsed = asObject(value);
const transport = asString(parsed.transport).trim();
@ -715,13 +768,24 @@ export async function prepareSandboxManagedRuntime(input: {
const runtimeRootDir = path.posix.join(workspaceRemoteDir, ".paperclip-runtime", input.adapterKey);
const syncWorkspace = input.syncWorkspace !== false;
// Wrap a host-side staging sub-step in its own span when the caller injects a
// runtime span runner. Mirrors the `pack` span below: the runner defaults to a
// no-op, so a caller with no injected runner keeps the current control flow,
// and a throwing runner never changes it (see `createRuntimeSpanRunner`). Each
// span parents under the `stage.sync` step, so the two pre-`pack` operations —
// the git enumeration and the baseline content-hash walk — stop showing up as
// a hidden gap at the head of the step.
// Reject any unsafe asset key before an archive path or an asset directory is
// built from it. This runs before the git snapshot work so a bad key fails fast.
for (const asset of input.assets ?? []) {
assertRuntimeAssetKeyIsSafe(asset.key);
}
// Wrap a host-side staging sub-step or one scheduler task in its own span when
// the caller injects a runtime span runner. The runner defaults to a no-op, so
// a caller with no injected runner keeps the current control flow, and a
// throwing runner never changes it (see `createRuntimeSpanRunner`). The two
// pre-`pack` operations — the git enumeration and the baseline content-hash
// walk — parent under the `stage.sync` step, so they stop showing up as a
// hidden gap at the head of the step. Each inbound task (`stage.workspace`,
// `stage.asset.<key>`, `stage.project.<id>`) and each outbound restore task
// (`restore.workspace`, `restore.asset.<key>`) opens its own span, so two
// concurrent tasks produce overlapping spans and the `pack` span nests under
// `stage.workspace`. The outbound spans parent under the run's `sandbox.syncBack`
// span, because the teardown runs the restore inside that span.
const runStepSpan = <T>(name: string, work: () => Promise<T>): Promise<T> =>
input.runtimeSpan ? input.runtimeSpan(name, work) : work();
@ -851,167 +915,192 @@ export async function prepareSandboxManagedRuntime(input: {
await upload.finish(params.progressBytes, params.progressBytes);
};
// Build the ordered inbound operation task list. Each task stages one inbound
// operation from start to end: it packs the bytes, confines the mappings, and
// uploads them. The shared scheduler starts the tasks. When the sync client
// permits concurrency, the scheduler keeps at most SYNC_OPERATION_CONCURRENCY_LIMIT
// tasks active. When the client does not permit concurrency, the scheduler runs
// one task at a time in this order. The scheduler settles every started task
// before it returns, so no upload outlives the coordinator and ACP never starts
// before the barrier.
const inboundTasks: Array<SyncOperationTask<void>> = [];
// A required-flag list, one entry per task and index-aligned with inboundTasks.
// The workspace and each asset are required operations: a rejection is fatal
// after the barrier. Each referenced project is a nonfatal operation: its task
// records its own failure and never rejects.
const inboundTaskIsRequired: boolean[] = [];
if (syncWorkspace) {
// A git-backed workspace and a plain workspace both stage through ONE
// confined `syncIn` operation. A git-backed workspace carries TWO host tars —
// the git-history clone and the working-tree overlay — as two `file` mappings
// on the SAME operation, with their extract commands as ordered
// `postUploadCommands`. One operation shares one mkdir, one confine guard, one
// `uploadFiles`, and one rename exec, so the second `syncIn` round trip is
// removed. Build the whole merged file set and command list BEFORE the confine
// guard runs (inside `stageConfinedSyncIn`); never append a mapping after it.
const workspaceFiles: SandboxSyncFileMapping[] = [];
const workspacePostUploadCommands: SandboxPostUploadCommand[] = [];
let workspaceUploadBytes = 0;
inboundTaskIsRequired.push(true);
inboundTasks.push(() =>
runStepSpan("stage.workspace", async () => {
// A git-backed workspace and a plain workspace both stage through ONE
// confined `syncIn` operation. A git-backed workspace carries TWO host tars —
// the git-history clone and the working-tree overlay — as two `file` mappings
// on the SAME operation, with their extract commands as ordered
// `postUploadCommands`. One operation shares one mkdir, one confine guard, one
// `uploadFiles`, and one rename exec, so the second `syncIn` round trip is
// removed. Build the whole merged file set and command list BEFORE the confine
// guard runs (inside `stageConfinedSyncIn`); never append a mapping after it.
const workspaceFiles: SandboxSyncFileMapping[] = [];
const workspacePostUploadCommands: SandboxPostUploadCommand[] = [];
let workspaceUploadBytes = 0;
// Build both host tarballs (the git-history tar and the workspace-overlay
// tar) inside one host span named `pack`. This span makes the host pack
// time visible under the `stage.sync` step, where it is otherwise a hidden
// gap with no span. The transfer (`stageConfinedSyncIn`) runs after this
// span, so `pack` measures only the host tar-build cost. The runner
// defaults to a no-op, so a caller with no injected runner keeps the
// current behavior and control flow.
await runStepSpan("pack", async () => {
// 1. git-history tar (git-backed workspace only). Both tar targets live under
// `runtimeRootDir` (`.paperclip-runtime/<adapterKey>`). The git extract
// wipes the target tree EXCEPT `.paperclip-runtime`, so the overlay tar,
// which sits under `.paperclip-runtime`, survives to run its own extract.
if (gitSnapshot) {
await emitRuntimeStatus(input.onRuntimeProgress, "git_sync", "Syncing git history to sandbox");
const gitTarPath = path.join(tempDir, "git-workspace.tar");
const remoteGitTar = path.posix.join(runtimeRootDir, "git-workspace-upload.tar");
await withShallowGitWorkspaceClone({
localDir: input.workspaceLocalDir,
snapshot: gitSnapshot,
}, async (cloneDir) => {
// Build both host tarballs (the git-history tar and the workspace-overlay
// tar) inside one host span named `pack`. This span makes the host pack
// time visible under the `stage.sync` step, where it is otherwise a hidden
// gap with no span. The transfer (`stageConfinedSyncIn`) runs after this
// span, so `pack` measures only the host tar-build cost. The runner
// defaults to a no-op, so a caller with no injected runner keeps the
// current behavior and control flow.
await runStepSpan("pack", async () => {
// 1. git-history tar (git-backed workspace only). Both tar targets live under
// `runtimeRootDir` (`.paperclip-runtime/<adapterKey>`). The git extract
// wipes the target tree EXCEPT `.paperclip-runtime`, so the overlay tar,
// which sits under `.paperclip-runtime`, survives to run its own extract.
if (gitSnapshot) {
await emitRuntimeStatus(input.onRuntimeProgress, "git_sync", "Syncing git history to sandbox");
const gitTarPath = path.join(tempDir, "git-workspace.tar");
const remoteGitTar = path.posix.join(runtimeRootDir, "git-workspace-upload.tar");
await withShallowGitWorkspaceClone({
localDir: input.workspaceLocalDir,
snapshot: gitSnapshot,
}, async (cloneDir) => {
await createTarballFromDirectory({
localDir: cloneDir,
archivePath: gitTarPath,
exclude: [".paperclip-runtime"],
});
});
workspaceFiles.push({ sourcePath: gitTarPath, targetPath: remoteGitTar, kind: "file", access: "rw", writablePath: workspaceRemoteDir });
workspacePostUploadCommands.push({
command: buildWorkspaceTarExtractCommand({
workspaceRemoteDir,
remoteTar: remoteGitTar,
wipeExceptNames: [".paperclip-runtime"],
}),
});
workspaceUploadBytes += (await fs.stat(gitTarPath)).size;
}
// 2. workspace-overlay tar. A git-backed overlay merges on top of the just
// extracted git tree (no wipe); a plain workspace wipes every child except
// the preserved names first. The extract runs AFTER the git extract.
await emitRuntimeStatus(input.onRuntimeProgress, "config_sync", "Syncing workspace to sandbox");
const workspaceTarPath = path.join(tempDir, "workspace.tar");
const workspaceArchiveDir = gitSnapshot ? path.join(tempDir, "workspace-overlay") : input.workspaceLocalDir;
if (gitSnapshot) {
await copySelectedWorkspaceEntries({
sourceDir: input.workspaceLocalDir,
targetDir: workspaceArchiveDir,
relativePaths: gitSnapshot.overlayPaths,
exclude: workspaceArchiveExclude,
});
}
await createTarballFromDirectory({
localDir: cloneDir,
archivePath: gitTarPath,
exclude: [".paperclip-runtime"],
localDir: workspaceArchiveDir,
archivePath: workspaceTarPath,
exclude: gitSnapshot ? undefined : workspaceArchiveExclude,
});
const remoteWorkspaceTar = path.posix.join(runtimeRootDir, "workspace-upload.tar");
workspaceFiles.push({ sourcePath: workspaceTarPath, targetPath: remoteWorkspaceTar, kind: "file", access: "rw", writablePath: workspaceRemoteDir });
workspacePostUploadCommands.push({
command: buildWorkspaceTarExtractCommand({
workspaceRemoteDir,
remoteTar: remoteWorkspaceTar,
wipeExceptNames: gitSnapshot ? null : [...preservedNames],
}),
});
// 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,
}),
});
}
workspaceUploadBytes += (await fs.stat(workspaceTarPath)).size;
});
workspaceFiles.push({ sourcePath: gitTarPath, targetPath: remoteGitTar, kind: "file", access: "rw", writablePath: workspaceRemoteDir });
workspacePostUploadCommands.push({
command: buildWorkspaceTarExtractCommand({
workspaceRemoteDir,
remoteTar: remoteGitTar,
wipeExceptNames: [".paperclip-runtime"],
}),
});
workspaceUploadBytes += (await fs.stat(gitTarPath)).size;
}
// 2. workspace-overlay tar. A git-backed overlay merges on top of the just
// extracted git tree (no wipe); a plain workspace wipes every child except
// the preserved names first. The extract runs AFTER the git extract.
await emitRuntimeStatus(input.onRuntimeProgress, "config_sync", "Syncing workspace to sandbox");
const workspaceTarPath = path.join(tempDir, "workspace.tar");
const workspaceArchiveDir = gitSnapshot ? path.join(tempDir, "workspace-overlay") : input.workspaceLocalDir;
if (gitSnapshot) {
await copySelectedWorkspaceEntries({
sourceDir: input.workspaceLocalDir,
targetDir: workspaceArchiveDir,
relativePaths: gitSnapshot.overlayPaths,
exclude: workspaceArchiveExclude,
// One confined `syncIn` for the whole merged workspace file set. The confine
// guard covers every mapping BEFORE any bytes upload (fail-closed): a source
// or target escape in EITHER tar mapping stops the upload of both.
await stageConfinedSyncIn({
files: workspaceFiles,
postUploadCommands: workspacePostUploadCommands,
sourceRoots: [tempDir],
targetRoots: [runtimeRootDir],
progressLabel: "workspace",
statusPhase: "config_sync",
progressBytes: workspaceUploadBytes,
});
}
await createTarballFromDirectory({
localDir: workspaceArchiveDir,
archivePath: workspaceTarPath,
exclude: gitSnapshot ? undefined : workspaceArchiveExclude,
});
const remoteWorkspaceTar = path.posix.join(runtimeRootDir, "workspace-upload.tar");
workspaceFiles.push({ sourcePath: workspaceTarPath, targetPath: remoteWorkspaceTar, kind: "file", access: "rw", writablePath: workspaceRemoteDir });
workspacePostUploadCommands.push({
command: buildWorkspaceTarExtractCommand({
workspaceRemoteDir,
remoteTar: remoteWorkspaceTar,
wipeExceptNames: gitSnapshot ? null : [...preservedNames],
}),
});
// 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,
}),
});
}
workspaceUploadBytes += (await fs.stat(workspaceTarPath)).size;
});
// One confined `syncIn` for the whole merged workspace file set. The confine
// guard covers every mapping BEFORE any bytes upload (fail-closed): a source
// or target escape in EITHER tar mapping stops the upload of both.
await stageConfinedSyncIn({
files: workspaceFiles,
postUploadCommands: workspacePostUploadCommands,
sourceRoots: [tempDir],
targetRoots: [runtimeRootDir],
progressLabel: "workspace",
statusPhase: "config_sync",
progressBytes: workspaceUploadBytes,
});
}),
);
}
for (const asset of input.assets ?? []) {
await emitRuntimeStatus(input.onRuntimeProgress, "config_sync", "Syncing runtime assets to sandbox");
const remoteAssetDir = path.posix.join(runtimeRootDir, asset.key);
const remoteAssetTar = path.posix.join(runtimeRootDir, `${asset.key}-upload.tar`);
// Every asset — default OR custom-provisioned (e.g. an adapter credential
// merge) — rides one `syncIn` operation: the asset tar plus any staged
// helper files as `files` mappings, and the extract/merge command as the
// ordered post-upload command. There is no native-diversion gate; a
// custom-provisioned asset's bytes now ride native `uploadFiles` and its
// command runs as a provider-executed post-upload command.
const assetTarPath = path.join(tempDir, `${asset.key}.tar`);
await createTarballFromDirectory({
localDir: asset.localDir,
archivePath: assetTarPath,
followSymlinks: asset.followSymlinks,
exclude: asset.exclude,
});
const files: SandboxSyncFileMapping[] = [
{ sourcePath: assetTarPath, targetPath: remoteAssetTar, kind: "file", access: "rw", writablePath: remoteAssetDir },
];
// Stage provision helper files (e.g. the merge scripts) into the temp dir
// and map them alongside the asset tar so they ride the same native upload.
for (const stageFile of asset.provision?.stageFiles ?? []) {
const safeName = stageFile.name;
if (/[\\/]|\.\.(\.|$)/.test(safeName) || safeName === "..") {
throw new Error(`provision stageFile.name must be a simple basename, got: ${safeName}`);
}
const stageBytes = typeof stageFile.contents === "string"
? Buffer.from(stageFile.contents)
: stageFile.contents;
const stageHostPath = path.join(tempDir, `${asset.key}.stage.${safeName}`);
await fs.writeFile(stageHostPath, stageBytes);
// A stage helper file (for example a merge script) is a read-only input
// that the provision command reads; the agent does not change it and does
// not keep it. So it is `access: "ro"` and never joins the writable set.
files.push({
sourcePath: stageHostPath,
targetPath: path.posix.join(runtimeRootDir, safeName),
kind: "file",
access: "ro",
});
}
const postUploadCommand = asset.provision?.postUploadCommand?.({
assetTarPath: remoteAssetTar,
assetDir: remoteAssetDir,
runtimeRootDir,
}) ?? buildDefaultExtractRuntimeAssetCommand({ remoteAssetDir, remoteAssetTar });
const assetTarSize = (await fs.stat(assetTarPath)).size;
await stageConfinedSyncIn({
files,
postUploadCommands: [{ command: postUploadCommand }],
sourceRoots: [tempDir],
targetRoots: [runtimeRootDir],
progressLabel: asset.key,
statusPhase: "config_sync",
progressBytes: assetTarSize,
});
inboundTaskIsRequired.push(true);
inboundTasks.push(() =>
runStepSpan(`stage.asset.${asset.key}`, async () => {
await emitRuntimeStatus(input.onRuntimeProgress, "config_sync", "Syncing runtime assets to sandbox");
const remoteAssetDir = path.posix.join(runtimeRootDir, asset.key);
const remoteAssetTar = path.posix.join(runtimeRootDir, `${asset.key}-upload.tar`);
// Every asset — default OR custom-provisioned (e.g. an adapter credential
// merge) — rides one `syncIn` operation: the asset tar plus any staged
// helper files as `files` mappings, and the extract/merge command as the
// ordered post-upload command. There is no native-diversion gate; a
// custom-provisioned asset's bytes now ride native `uploadFiles` and its
// command runs as a provider-executed post-upload command.
const assetTarPath = path.join(tempDir, `${asset.key}.tar`);
await createTarballFromDirectory({
localDir: asset.localDir,
archivePath: assetTarPath,
followSymlinks: asset.followSymlinks,
exclude: asset.exclude,
});
const files: SandboxSyncFileMapping[] = [
{ sourcePath: assetTarPath, targetPath: remoteAssetTar, kind: "file", access: "rw", writablePath: remoteAssetDir },
];
// Stage provision helper files (e.g. the merge scripts) into the temp dir
// and map them alongside the asset tar so they ride the same native upload.
for (const stageFile of asset.provision?.stageFiles ?? []) {
const safeName = stageFile.name;
if (/[\\/]|\.\.(\.|$)/.test(safeName) || safeName === "..") {
throw new Error(`provision stageFile.name must be a simple basename, got: ${safeName}`);
}
const stageBytes = typeof stageFile.contents === "string"
? Buffer.from(stageFile.contents)
: stageFile.contents;
const stageHostPath = path.join(tempDir, `${asset.key}.stage.${safeName}`);
await fs.writeFile(stageHostPath, stageBytes);
// A stage helper file (for example a merge script) is a read-only input
// that the provision command reads; the agent does not change it and does
// not keep it. So it is `access: "ro"` and never joins the writable set.
files.push({
sourcePath: stageHostPath,
targetPath: path.posix.join(runtimeRootDir, safeName),
kind: "file",
access: "ro",
});
}
const postUploadCommand = asset.provision?.postUploadCommand?.({
assetTarPath: remoteAssetTar,
assetDir: remoteAssetDir,
runtimeRootDir,
}) ?? buildDefaultExtractRuntimeAssetCommand({ remoteAssetDir, remoteAssetTar });
const assetTarSize = (await fs.stat(assetTarPath)).size;
await stageConfinedSyncIn({
files,
postUploadCommands: [{ command: postUploadCommand }],
sourceRoots: [tempDir],
targetRoots: [runtimeRootDir],
progressLabel: asset.key,
statusPhase: "config_sync",
progressBytes: assetTarSize,
});
}),
);
}
// Stage each referenced (additional) project as a plain, read-only tree in
@ -1023,46 +1112,83 @@ export async function prepareSandboxManagedRuntime(input: {
// project's confinement or sync failure logs a warning and is skipped, and
// the run plus the other projects continue. Only a project that stages
// successfully appears in `additionalSourceDirs`.
for (const source of input.additionalSources ?? []) {
const { localPath, projectId } = source;
const label = `project-${projectId}`;
try {
if (!path.posix.isAbsolute(localPath)) {
throw new Error(`additional source localPath is not an absolute path: ${localPath}`);
}
if (
projectId.length === 0 ||
projectId.includes("/") ||
projectId.includes("\\") ||
projectId.includes("..")
) {
throw new Error(`additional source projectId is not a simple path segment: ${projectId}`);
}
const remoteProjectDir = path.posix.join(runtimeRootDir, label);
await emitRuntimeStatus(input.onRuntimeProgress, "config_sync", "Syncing referenced project to sandbox");
await stageConfinedSyncIn({
files: [{
sourcePath: localPath,
targetPath: remoteProjectDir,
kind: "directory",
exclude: additionalSourceExclude,
access: "ro",
}],
sourceRoots: [localPath],
targetRoots: [remoteProjectDir],
progressLabel: label,
statusPhase: "config_sync",
progressBytes: 0,
});
additionalSourceDirs[projectId] = remoteProjectDir;
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
console.warn(
`[paperclip] Failed to stage referenced project ${projectId}; skipping it. ${message}`,
);
// Record the failure as a first-class per-project outcome so the run can count it in the
// requested-vs-synced accounting instead of losing it to a warning line.
additionalSourceFailures.push({ projectId, error: message });
const additionalSourceList = input.additionalSources ?? [];
// Record each referenced-project failure in its input-order slot. Slot order
// keeps the failure list stable no matter the task completion order under
// concurrency.
const additionalFailureSlots: Array<AdditionalSourceStagingFailure | null> =
additionalSourceList.map(() => null);
additionalSourceList.forEach((source, sourceIndex) => {
inboundTaskIsRequired.push(false);
inboundTasks.push(() =>
runStepSpan(`stage.project.${source.projectId}`, async () => {
const { localPath, projectId } = source;
const label = `project-${projectId}`;
try {
if (!path.posix.isAbsolute(localPath)) {
throw new Error(`additional source localPath is not an absolute path: ${localPath}`);
}
if (
projectId.length === 0 ||
projectId.includes("/") ||
projectId.includes("\\") ||
projectId.includes("..")
) {
throw new Error(`additional source projectId is not a simple path segment: ${projectId}`);
}
const remoteProjectDir = path.posix.join(runtimeRootDir, label);
await emitRuntimeStatus(input.onRuntimeProgress, "config_sync", "Syncing referenced project to sandbox");
await stageConfinedSyncIn({
files: [{
sourcePath: localPath,
targetPath: remoteProjectDir,
kind: "directory",
exclude: additionalSourceExclude,
access: "ro",
}],
sourceRoots: [localPath],
targetRoots: [remoteProjectDir],
progressLabel: label,
statusPhase: "config_sync",
progressBytes: 0,
});
additionalSourceDirs[projectId] = remoteProjectDir;
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
// Record the failure as a first-class per-project outcome so the run can count it in the
// requested-vs-synced accounting and surface the reason on the run result and the run log.
// The structured slot carries the reason, so no `console.warn` line is needed here.
additionalFailureSlots[sourceIndex] = { projectId, error: message };
}
}),
);
});
// Run every inbound operation through the shared scheduler. It starts the
// tasks under the client's concurrency permission and the bound, and it
// settles every started task before it returns. This is the startup barrier:
// the coordinator returns only after every upload settles.
const inboundResults = await scheduleSyncOperations(
inboundTasks,
input.client.allowConcurrentSyncOperations === true,
SYNC_OPERATION_CONCURRENCY_LIMIT,
);
// Collect the referenced-project failures in stable input order.
for (const failure of additionalFailureSlots) {
if (failure) {
additionalSourceFailures.push(failure);
}
}
// Select a fatal failure in stable operation order. The workspace comes first,
// then each asset. A referenced-project task never rejects, so a nonfatal
// outcome never appears here. Raise the first required rejection, so two
// required failures raise the earlier one.
for (let index = 0; index < inboundResults.length; index += 1) {
const result = inboundResults[index];
if (inboundTaskIsRequired[index] && result.status === "rejected") {
throw result.reason;
}
}
});
@ -1081,182 +1207,227 @@ export async function prepareSandboxManagedRuntime(input: {
additionalSourceFailures,
restoreWorkspace: async (onProgress?: RuntimeProgressSink) => {
const restoreSink = onProgress ?? input.onProgress;
if (!syncWorkspace) {
for (const asset of input.assets ?? []) {
if (!asset.restore) continue;
await asset.restore({
assetDir: path.posix.join(runtimeRootDir, asset.key),
readFile: async (remotePath) => toBuffer(await input.client.readFile(remotePath)),
});
}
return;
}
await withTempDir("paperclip-sandbox-restore-", async (tempDir) => {
let importedRef: string | null = null;
let importedHead: string | null = null;
let remoteWorkspaceStatus = "dirty";
try {
if (gitSnapshot) {
await emitRuntimeStatus(input.onRuntimeProgress, "export", "Exporting git changes from sandbox");
importedRef = createImportedGitRef("sandbox");
const remoteGitBundle = path.posix.join(runtimeRootDir, "git-delta.bundle");
const remoteWorkspaceStatusPath = path.posix.join(runtimeRootDir, "workspace-status.txt");
const exportRef = createRemoteGitExportRef("sandbox");
const localBundlePath = path.join(tempDir, "git-delta.bundle");
// Export the sandbox history and import it into the host workspace.
// The delta bundle assumes the host holds the bundle's boundary
// commit; when the host has been reset far enough that it does not,
// the import fails on a missing prerequisite. In that case re-export
// a full, self-contained bundle from the still-live sandbox rather
// than discard the completed run.
const exportAndImport = async (forceFullBundle: boolean): Promise<string> => {
await input.client.run(
`sh -c ${shellQuote(buildRemoteGitDeltaBundleScript({
remoteDir: workspaceRemoteDir,
baseSha: gitSnapshot.headCommit,
exportRef,
bundlePath: remoteGitBundle,
statusPath: forceFullBundle ? undefined : remoteWorkspaceStatusPath,
forceFullBundle,
}))}`,
{ timeoutMs: input.spec.timeoutMs },
);
const gitExport = makeTransferProgress(
restoreSink,
"Exporting git history",
"from",
undefined,
{ sink: input.onRuntimeProgress, phase: "export" },
);
const bundleBytes = await input.client.readFile(remoteGitBundle, gitExport.options);
const bundleBuffer = toBuffer(bundleBytes);
await gitExport.finish(bundleBuffer.byteLength, bundleBuffer.byteLength);
await input.client.remove(remoteGitBundle).catch(() => undefined);
if (!forceFullBundle) {
remoteWorkspaceStatus = await input.client.readFile(remoteWorkspaceStatusPath)
.then((bytes) => toBuffer(bytes).toString("utf8").trim())
.catch(() => "dirty");
remoteWorkspaceStatus = remoteWorkspaceStatus === "clean" ? "clean" : "dirty";
await input.client.remove(remoteWorkspaceStatusPath).catch(() => undefined);
// Build the ordered outbound restore task list. Each task runs one
// restore from start to end: the workspace restore, or one asset restore.
// The shared scheduler starts the tasks. When the sync client permits
// concurrency, the scheduler keeps at most SYNC_OPERATION_CONCURRENCY_LIMIT
// tasks active. When the client does not permit concurrency, the scheduler
// runs one task at a time in this order. The scheduler settles every
// started task before it returns, so lease teardown never starts while an
// outbound restore task still writes host data.
const outboundTasks: Array<SyncOperationTask<void>> = [];
// The workspace restore task runs only when the run syncs the workspace.
// The task exports the sandbox git history (git-backed workspace), reads
// the sandbox workspace back, and merges it into the host workspace root.
// The merge is the only outbound write inside the host workspace root.
// Every other task writes a disjoint host target: an asset restore writes
// its own store outside the workspace root. So the workspace task and the
// asset tasks share one parallel set. Keep any future asset that must write
// inside the host workspace root out of this set, and run it after the
// merge. Each task gets its own restore temp directory, so two concurrent
// tasks never share scratch state.
if (syncWorkspace) {
outboundTasks.push(() =>
runStepSpan("restore.workspace", async () => {
await withTempDir("paperclip-sandbox-restore-", async (tempDir) => {
let importedRef: string | null = null;
let importedHead: string | null = null;
let remoteWorkspaceStatus = "dirty";
try {
if (gitSnapshot) {
await emitRuntimeStatus(input.onRuntimeProgress, "export", "Exporting git changes from sandbox");
importedRef = createImportedGitRef("sandbox");
const remoteGitBundle = path.posix.join(runtimeRootDir, "git-delta.bundle");
const remoteWorkspaceStatusPath = path.posix.join(runtimeRootDir, "workspace-status.txt");
const exportRef = createRemoteGitExportRef("sandbox");
const localBundlePath = path.join(tempDir, "git-delta.bundle");
// Export the sandbox history and import it into the host workspace.
// The delta bundle assumes the host holds the bundle's boundary
// commit; when the host has been reset far enough that it does not,
// the import fails on a missing prerequisite. In that case re-export
// a full, self-contained bundle from the still-live sandbox rather
// than discard the completed run.
const exportAndImport = async (forceFullBundle: boolean): Promise<string> => {
await input.client.run(
`sh -c ${shellQuote(buildRemoteGitDeltaBundleScript({
remoteDir: workspaceRemoteDir,
baseSha: gitSnapshot.headCommit,
exportRef,
bundlePath: remoteGitBundle,
statusPath: forceFullBundle ? undefined : remoteWorkspaceStatusPath,
forceFullBundle,
}))}`,
{ timeoutMs: input.spec.timeoutMs },
);
const gitExport = makeTransferProgress(
restoreSink,
"Exporting git history",
"from",
undefined,
{ sink: input.onRuntimeProgress, phase: "export" },
);
const bundleBytes = await input.client.readFile(remoteGitBundle, gitExport.options);
const bundleBuffer = toBuffer(bundleBytes);
await gitExport.finish(bundleBuffer.byteLength, bundleBuffer.byteLength);
await input.client.remove(remoteGitBundle).catch(() => undefined);
if (!forceFullBundle) {
remoteWorkspaceStatus = await input.client.readFile(remoteWorkspaceStatusPath)
.then((bytes) => toBuffer(bytes).toString("utf8").trim())
.catch(() => "dirty");
remoteWorkspaceStatus = remoteWorkspaceStatus === "clean" ? "clean" : "dirty";
await input.client.remove(remoteWorkspaceStatusPath).catch(() => undefined);
}
await fs.writeFile(localBundlePath, bundleBuffer);
return fetchGitBundleIntoLocalRef({
localDir: input.workspaceLocalDir,
bundlePath: localBundlePath,
exportRef,
importedRef: importedRef!,
baseSha: gitSnapshot.headCommit,
});
};
try {
importedHead = await exportAndImport(false);
} catch (error) {
if (!isMissingGitPrerequisiteError(error)) throw error;
importedHead = await exportAndImport(true);
}
}
await emitRuntimeStatus(input.onRuntimeProgress, "restore", "Restoring workspace from sandbox");
const extractedDir = path.join(tempDir, "workspace");
if (nativeSyncOut) {
// Native outbound: the provider materializes the sandbox workspace into
// a fresh host directory. It is a clean destroy-then-replace into a
// temp dir the orchestrator just created, so it maps exactly to a
// generic directory file mapping; the host-side baseline merge below is
// unchanged.
const operations: SandboxSyncOperation[] = [{
operationId: nextSyncOperationId(),
files: [{
sourcePath: workspaceRemoteDir,
targetPath: extractedDir,
kind: "directory",
exclude: restoreExclude,
}],
}];
assertSyncOperationsConfined(operations, {
sourceRoots: [workspaceRemoteDir],
targetRoots: [extractedDir],
});
await fs.mkdir(extractedDir, { recursive: true });
const workspaceRestore = makeTransferProgress(
restoreSink,
"Restoring",
"from",
"workspace",
{ sink: input.onRuntimeProgress, phase: "restore" },
);
await input.client.syncOut!(operations);
await workspaceRestore.finish(0, 0);
} else {
const remoteWorkspaceTar = path.posix.join(runtimeRootDir, "workspace-download.tar");
await input.client.run(
`sh -c ${shellQuote(createRemoteTarballFromDirectoryCommand({
remoteDir: workspaceRemoteDir,
archivePath: remoteWorkspaceTar,
exclude: restoreExclude,
}))}`,
{ timeoutMs: input.spec.timeoutMs },
);
const workspaceRestore = makeTransferProgress(
restoreSink,
"Restoring",
"from",
"workspace",
{ sink: input.onRuntimeProgress, phase: "restore" },
);
const archiveBytes = await input.client.readFile(remoteWorkspaceTar, workspaceRestore.options);
const archiveBuffer = toBuffer(archiveBytes);
await workspaceRestore.finish(archiveBuffer.byteLength, archiveBuffer.byteLength);
await input.client.remove(remoteWorkspaceTar).catch(() => undefined);
const localArchivePath = path.join(tempDir, "workspace.tar");
await fs.writeFile(localArchivePath, archiveBuffer);
await extractTarballToDirectory({
archivePath: localArchivePath,
localDir: extractedDir,
});
}
const gitHeadToIntegrate = importedHead;
await mergeDirectoryWithBaseline({
baseline: baselineSnapshot!,
sourceDir: extractedDir,
targetDir: input.workspaceLocalDir,
beforeApply: gitHeadToIntegrate
? async () => {
await integrateImportedGitHead({
localDir: input.workspaceLocalDir,
importedHead: gitHeadToIntegrate,
});
}
: undefined,
afterApply: gitSnapshot
? async () => {
await resetLocalGitIndexToHead({
localDir: input.workspaceLocalDir,
checkWorkingTreeClean: remoteWorkspaceStatus === "clean",
});
}
: undefined,
});
} finally {
await emitRuntimeStatus(input.onRuntimeProgress, "finalize", "Finalizing sandbox workspace");
if (importedRef) {
await deleteLocalGitRef({ localDir: input.workspaceLocalDir, ref: importedRef });
}
}
await fs.writeFile(localBundlePath, bundleBuffer);
return fetchGitBundleIntoLocalRef({
localDir: input.workspaceLocalDir,
bundlePath: localBundlePath,
exportRef,
importedRef: importedRef!,
baseSha: gitSnapshot.headCommit,
});
}),
);
}
// One restore task per asset that has a teardown contribution. Each task
// reads its asset back through `readFile` and writes a disjoint host
// target. Each task gets its own restore temp directory.
for (const asset of input.assets ?? []) {
if (!asset.restore) continue;
const assetRestore = asset.restore;
const assetKey = asset.key;
outboundTasks.push(() =>
runStepSpan(`restore.asset.${assetKey}`, async () => {
await withTempDir("paperclip-sandbox-restore-", async (tempDir) => {
await assetRestore({
assetDir: path.posix.join(runtimeRootDir, assetKey),
readFile: async (remotePath) => toBuffer(await input.client.readFile(remotePath)),
tempDir,
});
};
try {
importedHead = await exportAndImport(false);
} catch (error) {
if (!isMissingGitPrerequisiteError(error)) throw error;
importedHead = await exportAndImport(true);
}
}
await emitRuntimeStatus(input.onRuntimeProgress, "restore", "Restoring workspace from sandbox");
const extractedDir = path.join(tempDir, "workspace");
if (nativeSyncOut) {
// Native outbound: the provider materializes the sandbox workspace into
// a fresh host directory. It is a clean destroy-then-replace into a
// temp dir the orchestrator just created, so it maps exactly to a
// generic directory file mapping; the host-side baseline merge below is
// unchanged.
const operations: SandboxSyncOperation[] = [{
operationId: nextSyncOperationId(),
files: [{
sourcePath: workspaceRemoteDir,
targetPath: extractedDir,
kind: "directory",
exclude: restoreExclude,
}],
}];
assertSyncOperationsConfined(operations, {
sourceRoots: [workspaceRemoteDir],
targetRoots: [extractedDir],
});
await fs.mkdir(extractedDir, { recursive: true });
const workspaceRestore = makeTransferProgress(
restoreSink,
"Restoring",
"from",
"workspace",
{ sink: input.onRuntimeProgress, phase: "restore" },
);
await input.client.syncOut!(operations);
await workspaceRestore.finish(0, 0);
} else {
const remoteWorkspaceTar = path.posix.join(runtimeRootDir, "workspace-download.tar");
await input.client.run(
`sh -c ${shellQuote(createRemoteTarballFromDirectoryCommand({
remoteDir: workspaceRemoteDir,
archivePath: remoteWorkspaceTar,
exclude: restoreExclude,
}))}`,
{ timeoutMs: input.spec.timeoutMs },
);
const workspaceRestore = makeTransferProgress(
restoreSink,
"Restoring",
"from",
"workspace",
{ sink: input.onRuntimeProgress, phase: "restore" },
);
const archiveBytes = await input.client.readFile(remoteWorkspaceTar, workspaceRestore.options);
const archiveBuffer = toBuffer(archiveBytes);
await workspaceRestore.finish(archiveBuffer.byteLength, archiveBuffer.byteLength);
await input.client.remove(remoteWorkspaceTar).catch(() => undefined);
const localArchivePath = path.join(tempDir, "workspace.tar");
await fs.writeFile(localArchivePath, archiveBuffer);
await extractTarballToDirectory({
archivePath: localArchivePath,
localDir: extractedDir,
});
}
const gitHeadToIntegrate = importedHead;
await mergeDirectoryWithBaseline({
baseline: baselineSnapshot!,
sourceDir: extractedDir,
targetDir: input.workspaceLocalDir,
beforeApply: gitHeadToIntegrate
? async () => {
await integrateImportedGitHead({
localDir: input.workspaceLocalDir,
importedHead: gitHeadToIntegrate,
});
}
: undefined,
afterApply: gitSnapshot
? async () => {
await resetLocalGitIndexToHead({
localDir: input.workspaceLocalDir,
checkWorkingTreeClean: remoteWorkspaceStatus === "clean",
});
}
: undefined,
});
}),
);
}
// Per-asset teardown/outbound contributions. Generic: an asset with
// no `restore` is a no-op. The contribution reads back from the
// sandbox (e.g. a refreshed credential) via the provided `readFile`.
for (const asset of input.assets ?? []) {
if (!asset.restore) continue;
await asset.restore({
assetDir: path.posix.join(runtimeRootDir, asset.key),
readFile: async (remotePath) => toBuffer(await input.client.readFile(remotePath)),
});
}
} finally {
await emitRuntimeStatus(input.onRuntimeProgress, "finalize", "Finalizing sandbox workspace");
if (importedRef) {
await deleteLocalGitRef({ localDir: input.workspaceLocalDir, ref: importedRef });
}
// Run every outbound restore through the shared scheduler. It starts the
// tasks under the client concurrency permission and the bound, and it
// settles every started task before it returns. This is the teardown
// barrier.
const outboundResults = await scheduleSyncOperations(
outboundTasks,
input.client.allowConcurrentSyncOperations === true,
SYNC_OPERATION_CONCURRENCY_LIMIT,
);
// Every outbound restore is a required operation: a rejection is fatal.
// The workspace comes first, then each asset in order. Raise the first
// rejection in stable task order, so two failures raise the earlier one.
for (const result of outboundResults) {
if (result.status === "rejected") {
throw result.reason;
}
});
}
},
};
}

166
packages/adapter-utils/src/sync-operation-schedule.test.ts Normal file
View File

@ -0,0 +1,166 @@
import { describe, expect, it } from "vitest";
import {
SYNC_OPERATION_CONCURRENCY_LIMIT,
scheduleSyncOperations,
} from "./sync-operation-schedule.js";
// A deferred task fake. The `task` thunk records that it started and returns a
// promise that the test resolves or rejects by hand. The fake lets a test hold
// a task open and check the exact moment a later task starts.
interface DeferredTask<T> {
readonly task: () => Promise<T>;
resolve(value: T): void;
reject(reason: unknown): void;
started(): boolean;
}
function makeDeferred<T>(): DeferredTask<T> {
let started = false;
let resolveFn!: (value: T) => void;
let rejectFn!: (reason: unknown) => void;
const promise = new Promise<T>((resolve, reject) => {
resolveFn = resolve;
rejectFn = reject;
});
return {
task: () => {
started = true;
return promise;
},
resolve: (value: T) => resolveFn(value),
reject: (reason: unknown) => rejectFn(reason),
started: () => started,
};
}
// Drain the microtask and timer queues so every pending worker step runs.
function flush(): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, 0));
}
describe("scheduleSyncOperations", () => {
it("exposes the shared bound constant as 4", () => {
expect(SYNC_OPERATION_CONCURRENCY_LIMIT).toBe(4);
});
it("keeps at most `bound` tasks active in concurrent mode", async () => {
const deferreds = [makeDeferred<number>(), makeDeferred<number>(), makeDeferred<number>()];
const tasks = deferreds.map((deferred) => deferred.task);
const scheduled = scheduleSyncOperations(tasks, true, 2);
await flush();
// The bound is 2, so only the first two tasks start.
expect(deferreds[0].started()).toBe(true);
expect(deferreds[1].started()).toBe(true);
expect(deferreds[2].started()).toBe(false);
// One active task settles. A worker frees, so the third task starts.
deferreds[0].resolve(0);
await flush();
expect(deferreds[2].started()).toBe(true);
deferreds[1].resolve(1);
deferreds[2].resolve(2);
const results = await scheduled;
expect(results).toEqual([
{ status: "fulfilled", value: 0 },
{ status: "fulfilled", value: 1 },
{ status: "fulfilled", value: 2 },
]);
});
it("runs one task at a time in input order in serial mode", async () => {
const deferreds = [makeDeferred<number>(), makeDeferred<number>(), makeDeferred<number>()];
const tasks = deferreds.map((deferred) => deferred.task);
const scheduled = scheduleSyncOperations(tasks, false);
await flush();
// Serial mode holds one task active, so only the first task starts.
expect(deferreds[0].started()).toBe(true);
expect(deferreds[1].started()).toBe(false);
expect(deferreds[2].started()).toBe(false);
deferreds[0].resolve(0);
await flush();
// The next task starts only after the previous task settles.
expect(deferreds[1].started()).toBe(true);
expect(deferreds[2].started()).toBe(false);
deferreds[1].resolve(1);
await flush();
expect(deferreds[2].started()).toBe(true);
deferreds[2].resolve(2);
const results = await scheduled;
expect(results).toEqual([
{ status: "fulfilled", value: 0 },
{ status: "fulfilled", value: 1 },
{ status: "fulfilled", value: 2 },
]);
});
it("waits for every started task to settle when one rejects", async () => {
const first = makeDeferred<number>();
const second = makeDeferred<number>();
const tasks = [first.task, second.task];
const scheduled = scheduleSyncOperations(tasks, true, 2);
let settled = false;
void scheduled.then(() => {
settled = true;
});
await flush();
// Both tasks are active under the bound of 2.
expect(first.started()).toBe(true);
expect(second.started()).toBe(true);
// The first task rejects, but the second task stays open.
first.reject(new Error("first failed"));
await flush();
// The scheduler does not return before every started task settles.
expect(settled).toBe(false);
second.resolve(2);
const results = await scheduled;
expect(settled).toBe(true);
// The results keep input order, and the rejection carries its reason.
expect(results[0]).toEqual({ status: "rejected", reason: new Error("first failed") });
expect(results[1]).toEqual({ status: "fulfilled", value: 2 });
});
// One table proves settle-all and input-order for both call modes. Both future
// call sites share these proven cases.
const MODE_TABLE = [
{ name: "serial mode", concurrent: false, bound: SYNC_OPERATION_CONCURRENCY_LIMIT },
{ name: "concurrent mode", concurrent: true, bound: 2 },
];
for (const mode of MODE_TABLE) {
it(`returns settled results in input order in ${mode.name}`, async () => {
// The tasks settle out of order: index 2 first, then index 0, then a
// rejection at index 1. The result array must still keep input order.
const failure = new Error("index one failed");
const tasks: Array<() => Promise<string>> = [
() => Promise.resolve("zero"),
() => Promise.reject(failure),
() => Promise.resolve("two"),
];
const results = await scheduleSyncOperations(tasks, mode.concurrent, mode.bound);
expect(results).toEqual([
{ status: "fulfilled", value: "zero" },
{ status: "rejected", reason: failure },
{ status: "fulfilled", value: "two" },
]);
});
}
it("returns an empty result list for no tasks", async () => {
const results = await scheduleSyncOperations<number>([], true, 4);
expect(results).toEqual([]);
});
});

79
packages/adapter-utils/src/sync-operation-schedule.ts Normal file
View File

@ -0,0 +1,79 @@
// A direction-agnostic scheduler for file-sync operations. Both sync directions
// share this one function. The scheduler starts operation tasks under a bound,
// keeps serial order when concurrency is off, and settles every started task
// before it returns. It stays pure: it imports no runtime or provider module.
// The maximum number of active sync operations when concurrency is on. A caller
// passes this as the bound. Four keeps a useful parallel width without a burst
// of open file handles.
export const SYNC_OPERATION_CONCURRENCY_LIMIT = 4;
// A sync operation. The scheduler calls the thunk to start the operation, so the
// scheduler controls the exact start time and can hold the bound.
export type SyncOperationTask<T> = () => Promise<T>;
// Resolve the number of active tasks the scheduler allows.
// Serial mode allows one active task. Concurrent mode allows the bound, with a
// floor of one so a bad bound never stalls the scheduler.
function resolveActiveLimit(concurrent: boolean, bound: number): number {
if (!concurrent) {
return 1;
}
const flooredBound = Math.floor(bound);
if (!Number.isFinite(flooredBound) || flooredBound < 1) {
return 1;
}
return flooredBound;
}
/**
* Run an ordered list of sync operation tasks and settle every started task.
*
* When `concurrent` is false, the scheduler runs the tasks one at a time in
* input order. When `concurrent` is true, the scheduler keeps at most `bound`
* tasks active. The scheduler always waits for every started task to settle
* before it returns. It returns the settled results in input order.
*
* @param tasks Ordered task list. The scheduler calls each thunk to start it.
* @param concurrent Turn concurrency on or off.
* @param bound Maximum active tasks when `concurrent` is true.
* @returns Settled results in input order, one per task.
*/
export async function scheduleSyncOperations<T>(
tasks: ReadonlyArray<SyncOperationTask<T>>,
concurrent: boolean,
bound: number = SYNC_OPERATION_CONCURRENCY_LIMIT,
): Promise<Array<PromiseSettledResult<T>>> {
const results = new Array<PromiseSettledResult<T>>(tasks.length);
const activeLimit = resolveActiveLimit(concurrent, bound);
// A shared cursor over the input list. Each worker takes the next task in
// input order. The cursor keeps serial order and holds the active bound.
let nextIndex = 0;
// One worker settles tasks until the list is empty. A worker catches its own
// task rejection, so a rejection never stops the other workers and never
// rejects the scheduler. This guarantees settle-all.
async function runWorker(): Promise<void> {
while (nextIndex < tasks.length) {
const currentIndex = nextIndex;
nextIndex += 1;
try {
const value = await tasks[currentIndex]();
results[currentIndex] = { status: "fulfilled", value };
} catch (reason) {
results[currentIndex] = { status: "rejected", reason };
}
}
}
// Start at most `activeLimit` workers, and never more than the task count.
const workerCount = Math.min(activeLimit, tasks.length);
const workers: Array<Promise<void>> = [];
for (let worker = 0; worker < workerCount; worker += 1) {
workers.push(runWorker());
}
await Promise.all(workers);
return results;
}

7
packages/adapter-utils/src/types.ts
View File

@ -113,10 +113,11 @@ export interface AdapterExecutionResult {
* Each referenced (mentioned) project that failed to stage into the remote sandbox for this run,
* by `projectId`. The run continues without a failed project (per-project failure isolation); this
* field carries the failure back so the server counts it in the requested-vs-synced observability
* instead of losing it to a warning line. Absent or empty on a local target, or when every staged
* referenced project succeeded.
* instead of losing it to a warning line. Each entry pairs the `projectId` with the failure
* `error`, so a reader of the run learns why the project dropped. Absent or empty on a local
* target, or when every staged referenced project succeeded.
*/
referencedProjectStagingFailures?: Array<{ projectId: string }>;
referencedProjectStagingFailures?: Array<{ projectId: string; error: string }>;
summary?: string | null;
clearSession?: boolean;
question?: {

155
packages/plugins/sandbox-providers/daytona/src/file-sync.test.ts Normal file
View File

@ -0,0 +1,155 @@
import { promises as fs } from "node:fs";
import os from "node:os";
import path from "node:path";
import { afterEach, describe, expect, it } from "vitest";
// The plugin module imports `@daytonaio/sdk` as a value, but the sync tests never
// touch a real Daytona client — every sandbox call goes through a local mock. Stub
// the SDK so the import resolves without the excluded provider package.
import { vi } from "vitest";
vi.mock("@daytonaio/sdk", () => ({
Daytona: class MockDaytona {},
DaytonaNotFoundError: class MockDaytonaNotFoundError extends Error {},
DaytonaTimeoutError: class MockDaytonaTimeoutError extends Error {},
}));
import { performSyncIn } from "./file-sync.js";
import type { PluginSyncOperation } from "@paperclipai/plugin-sdk";
// One recorded in-sandbox command, so a test can assert the exact cleanup command.
interface RecordedCommand {
command: string;
}
// Build a mock Daytona sandbox for the inbound directory path. `executeCommand`
// records every command and returns exit 0, except that any command whose text
// matches `failCommandMatch` returns exit 1 (to simulate an extract failure).
// `uploadFiles` records each upload destination so a test can read the reserved
// scratch tar path the runtime chose.
function createMockSandbox(input: {
failCommandMatch?: RegExp;
uploadedDestinations: string[];
commands: RecordedCommand[];
}) {
return {
process: {
executeCommand: async (command: string) => {
input.commands.push({ command });
if (input.failCommandMatch && input.failCommandMatch.test(command)) {
return { exitCode: 1, result: "simulated extract failure" };
}
return { exitCode: 0, result: "" };
},
},
fs: {
uploadFiles: async (uploads: Array<{ source: string; destination: string }>) => {
for (const upload of uploads) input.uploadedDestinations.push(upload.destination);
},
},
};
}
describe("daytona file-sync inbound scratch cleanup", () => {
const cleanupDirs: string[] = [];
afterEach(async () => {
while (cleanupDirs.length > 0) {
const dir = cleanupDirs.pop();
if (!dir) continue;
await fs.rm(dir, { recursive: true, force: true }).catch(() => undefined);
}
});
it("removes the reserved scratch tar when a directory extraction fails", async () => {
const rootDir = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-daytona-scratch-"));
cleanupDirs.push(rootDir);
const sourceDir = path.join(rootDir, "referenced-project");
await fs.mkdir(sourceDir, { recursive: true });
await fs.writeFile(path.join(sourceDir, "README.md"), "referenced project\n", "utf8");
const remoteDir = "/workspace";
const targetPath = "/workspace/.paperclip-runtime/test-adapter/project-abc";
const uploadedDestinations: string[] = [];
const commands: RecordedCommand[] = [];
// Fail the extract round trip (the only command that runs `tar -xf`).
const sandbox = createMockSandbox({
failCommandMatch: /tar -xf/,
uploadedDestinations,
commands,
});
const operations: PluginSyncOperation[] = [{
operationId: "sync-op-1",
files: [{ sourcePath: sourceDir, targetPath, kind: "directory" }],
}];
await expect(
performSyncIn({
// The mock stands in for the Daytona SDK Sandbox; only the two methods the
// inbound directory path calls are needed.
sandbox: sandbox as never,
operations,
remoteDir,
timeoutSeconds: 30,
}),
).rejects.toThrow(/syncIn extract/);
// The runtime uploaded exactly one reserved scratch tar under the workspace
// root. Its name carries the reserved `.paperclip-upload-` prefix.
expect(uploadedDestinations).toHaveLength(1);
const scratchTar = uploadedDestinations[0];
expect(scratchTar).toContain(".paperclip-upload-");
expect(scratchTar.startsWith(`${remoteDir}/`)).toBe(true);
// The failure path swept the scratch tar: a standalone `rm -f` of the exact
// scratch path ran after the failed extract. The extract command itself also
// contains an `rm -f`, so the cleanup is the `rm -f` command that does NOT run
// `tar -xf`.
const cleanupCommands = commands.filter(
(entry) =>
entry.command.includes(scratchTar) &&
entry.command.includes("rm -f") &&
!entry.command.includes("tar -xf"),
);
expect(cleanupCommands.length).toBeGreaterThan(0);
});
it("does not sweep scratch on the happy path (extract removes it)", async () => {
const rootDir = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-daytona-scratch-"));
cleanupDirs.push(rootDir);
const sourceDir = path.join(rootDir, "referenced-project");
await fs.mkdir(sourceDir, { recursive: true });
await fs.writeFile(path.join(sourceDir, "README.md"), "referenced project\n", "utf8");
const remoteDir = "/workspace";
const targetPath = "/workspace/.paperclip-runtime/test-adapter/project-abc";
const uploadedDestinations: string[] = [];
const commands: RecordedCommand[] = [];
// No failure: every command succeeds, so the extract's own `rm -f` clears the
// scratch and no extra cleanup round trip runs.
const sandbox = createMockSandbox({ uploadedDestinations, commands });
const operations: PluginSyncOperation[] = [{
operationId: "sync-op-1",
files: [{ sourcePath: sourceDir, targetPath, kind: "directory" }],
}];
await performSyncIn({
sandbox: sandbox as never,
operations,
remoteDir,
timeoutSeconds: 30,
});
const scratchTar = uploadedDestinations[0];
// The standalone cleanup command (a `rm -f` without `tar -xf`) never runs on the
// happy path — only the extract command, which ends with its own `rm -f`.
const standaloneRemoves = commands.filter(
(entry) =>
entry.command.includes(scratchTar) &&
entry.command.includes("rm -f") &&
!entry.command.includes("tar -xf"),
);
expect(standaloneRemoves).toHaveLength(0);
});
});

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

@ -24,6 +24,10 @@ const SPAN_ATTR = {
packWallMs: `${SPAN_ATTR_PREFIX}pack.wall_ms`,
transferWallMs: `${SPAN_ATTR_PREFIX}transfer.wall_ms`,
transferGuardCount: `${SPAN_ATTR_PREFIX}transfer.guard.count`,
// The transfer direction: `inbound` for an upload to the sandbox, `outbound`
// for a download from the sandbox. Operation identity comes from the parent
// span, so the transfer span never carries an operation label.
transferDirection: `${SPAN_ATTR_PREFIX}transfer.direction`,
} as const;
/** The value of `SpanStatusCode.ERROR` in `@opentelemetry/api`. The plugin stays
@ -532,7 +536,10 @@ async function syncInFileMappings(input: {
await withProviderSpan({
name: "transfer",
wallMsAttr: SPAN_ATTR.transferWallMs,
attributes: { [SPAN_ATTR.transferGuardCount]: guardRoundTrips },
attributes: {
[SPAN_ATTR.transferGuardCount]: guardRoundTrips,
[SPAN_ATTR.transferDirection]: "inbound",
},
run: () => sandbox.fs.uploadFiles(uploads, timeoutSeconds),
});
@ -655,52 +662,67 @@ async function syncInDirectoryMapping(input: {
}),
});
guardRoundTrips += 1;
// `transfer` span: the real byte upload — `sandbox.fs.uploadFiles`.
await withProviderSpan({
name: "transfer",
wallMsAttr: SPAN_ATTR.transferWallMs,
attributes: { [SPAN_ATTR.transferGuardCount]: guardRoundTrips },
run: () =>
sandbox.fs.uploadFiles([{ source: archivePath, destination: remoteTar }], timeoutSeconds),
});
// Bind validation and extraction into ONE sandbox invocation, then extract into
// an OPEN directory inode rather than a path string. `exec 9<"$_pc_real"` itself
// walks every ancestor of `$_pc_real` during the `open()` syscall, so a sandbox
// process that swaps an ancestor component for a symlink AFTER `_pc_resolve`
// returns but BEFORE the `open()` resolves would leave fd 9 pointing at a
// directory outside the workspace — the earlier `case` check on the resolved
// string cannot see that. Close the gap with open-then-verify: open fd 9 (which
// PINS whatever inode `open()` landed on), then re-canonicalize `/proc/self/fd/9`
// — the pinned inode's own path — and confirm it is still inside `$_pc_root`
// before extracting. If an ancestor swap redirected the open, the pinned inode
// resolves outside the root and the verify fails closed (exit 42); once the
// verify passes, the inode is fixed and `tar -C /proc/self/fd/9` chdir's through
// the magic symlink to that exact inode, so a post-open ancestor swap cannot
// redirect the write. (The initial `case` on `$_pc_real` still fails fast on a
// pre-open escape; the fd re-verify is what makes the guarantee race-free.)
const extractScript = [
...canonicalizerPreamble(shellQuote(remoteDir)),
`_pc_real=$(_pc_resolve ${shellQuote(mapping.targetPath)}) || { echo "ESCAPE"; exit 42; };`,
`case "$_pc_real/" in "$_pc_root"/*) : ;; *) echo "ESCAPE"; exit 42 ;; esac;`,
`exec 9<"$_pc_real" || { echo "open failed"; exit 46; };`,
`_pc_fd_real=$(_pc_resolve /proc/self/fd/9) || { echo "ESCAPE"; exit 42; };`,
`case "$_pc_fd_real/" in "$_pc_root"/*) : ;; *) echo "ESCAPE"; exit 42 ;; esac;`,
`tar -xf ${shellQuote(remoteTar)} -C /proc/self/fd/9 || { echo "extract failed"; exit 43; };`,
`exec 9>&-;`,
`rm -f ${shellQuote(remoteTar)};`,
].join("\n");
// `extractTarball` span: one round trip — re-check the path, `tar -xf`, and
// remove the scratch tarball.
await withProviderSpan({
name: "extractTarball",
run: () =>
assertSandboxCommandOk(
sandbox,
`sh -c ${shellQuote(extractScript)}`,
timeoutSeconds,
"syncIn extract",
),
});
// The uploaded scratch tar lands at the workspace root as a reserved
// `.paperclip-upload-*` entry. The extract script below removes it only on
// success. On an upload or extract failure the scratch tar can remain, and the
// runtime workspace wipe preserves every `.paperclip-upload-*` entry, so a
// stale tar would surface in the agent workspace. Sweep the scratch on any
// failure — symmetric with the file-mapping path — so a failed sync (for
// example a referenced-project extraction) leaves no residue.
try {
// `transfer` span: the real byte upload — `sandbox.fs.uploadFiles`.
await withProviderSpan({
name: "transfer",
wallMsAttr: SPAN_ATTR.transferWallMs,
attributes: {
[SPAN_ATTR.transferGuardCount]: guardRoundTrips,
[SPAN_ATTR.transferDirection]: "inbound",
},
run: () =>
sandbox.fs.uploadFiles([{ source: archivePath, destination: remoteTar }], timeoutSeconds),
});
// Bind validation and extraction into ONE sandbox invocation, then extract into
// an OPEN directory inode rather than a path string. `exec 9<"$_pc_real"` itself
// walks every ancestor of `$_pc_real` during the `open()` syscall, so a sandbox
// process that swaps an ancestor component for a symlink AFTER `_pc_resolve`
// returns but BEFORE the `open()` resolves would leave fd 9 pointing at a
// directory outside the workspace — the earlier `case` check on the resolved
// string cannot see that. Close the gap with open-then-verify: open fd 9 (which
// PINS whatever inode `open()` landed on), then re-canonicalize `/proc/self/fd/9`
// — the pinned inode's own path — and confirm it is still inside `$_pc_root`
// before extracting. If an ancestor swap redirected the open, the pinned inode
// resolves outside the root and the verify fails closed (exit 42); once the
// verify passes, the inode is fixed and `tar -C /proc/self/fd/9` chdir's through
// the magic symlink to that exact inode, so a post-open ancestor swap cannot
// redirect the write. (The initial `case` on `$_pc_real` still fails fast on a
// pre-open escape; the fd re-verify is what makes the guarantee race-free.)
const extractScript = [
...canonicalizerPreamble(shellQuote(remoteDir)),
`_pc_real=$(_pc_resolve ${shellQuote(mapping.targetPath)}) || { echo "ESCAPE"; exit 42; };`,
`case "$_pc_real/" in "$_pc_root"/*) : ;; *) echo "ESCAPE"; exit 42 ;; esac;`,
`exec 9<"$_pc_real" || { echo "open failed"; exit 46; };`,
`_pc_fd_real=$(_pc_resolve /proc/self/fd/9) || { echo "ESCAPE"; exit 42; };`,
`case "$_pc_fd_real/" in "$_pc_root"/*) : ;; *) echo "ESCAPE"; exit 42 ;; esac;`,
`tar -xf ${shellQuote(remoteTar)} -C /proc/self/fd/9 || { echo "extract failed"; exit 43; };`,
`exec 9>&-;`,
`rm -f ${shellQuote(remoteTar)};`,
].join("\n");
// `extractTarball` span: one round trip — re-check the path, `tar -xf`, and
// remove the scratch tarball.
await withProviderSpan({
name: "extractTarball",
run: () =>
assertSandboxCommandOk(
sandbox,
`sh -c ${shellQuote(extractScript)}`,
timeoutSeconds,
"syncIn extract",
),
});
} catch (error) {
await removeSandboxScratch(sandbox, [remoteTar], timeoutSeconds);
throw error;
}
const filesTransferred = await countHostFiles(mapping.sourcePath, mapping.exclude);
return { filesTransferred, bytesTransferred };
});
@ -873,10 +895,25 @@ async function syncOutFileMappings(input: {
throw error;
}
// Count the serial sandbox round trips before the transfer, so the transfer
// span records how much of the wall time is guard cost. The validate-and-
// snapshot step is one sandbox round trip. This is symmetric with the inbound
// transfer span.
const guardRoundTrips = 1;
let responses: FileDownloadResponse[];
try {
// One batched bulk download for all file mappings, reading the snapshots.
responses = await sandbox.fs.downloadFiles(requests, timeoutSeconds);
// `transfer` span: the real byte download — `sandbox.fs.downloadFiles`.
responses = await withProviderSpan({
name: "transfer",
wallMsAttr: SPAN_ATTR.transferWallMs,
attributes: {
[SPAN_ATTR.transferGuardCount]: guardRoundTrips,
[SPAN_ATTR.transferDirection]: "outbound",
},
run: () => sandbox.fs.downloadFiles(requests, timeoutSeconds),
});
} catch (error) {
await cleanup();
throw error;
@ -926,6 +963,9 @@ async function syncOutDirectoryMapping(input: {
}): Promise<{ filesTransferred: number; bytesTransferred: number }> {
const { sandbox, mapping, remoteDir, timeoutSeconds } = input;
assertConfinedSandboxPath(remoteDir, mapping.sourcePath, "source");
// Count the serial sandbox round trips before the transfer, so the transfer
// span records how much of the wall time is guard cost.
let guardRoundTrips = 0;
await assertSandboxPathsConfined({
sandbox,
remoteDir,
@ -933,6 +973,7 @@ async function syncOutDirectoryMapping(input: {
timeoutSeconds,
label: "outbound symlink-escape guard",
});
guardRoundTrips += 1;
return withHostTempDir(async (tmp) => {
const remoteTar = path.posix.join(remoteDir, scratchName(".tar"));
@ -951,14 +992,22 @@ async function syncOutDirectoryMapping(input: {
`else tar -c --no-xattrs ${mapping.followSymlinks ? "-h " : ""}${excludeFlags} -f ${shellQuote(remoteTar)} -- "$@"; fi`,
].join(" && ");
await assertSandboxCommandOk(sandbox, `sh -c ${shellQuote(tarScript)}`, timeoutSeconds, "syncOut tar");
guardRoundTrips += 1;
const localTar = path.join(tmp, "sync-out.tar");
let bytesTransferred = 0;
try {
const responses = await sandbox.fs.downloadFiles(
[{ source: remoteTar, destination: localTar }],
timeoutSeconds,
);
// `transfer` span: the real byte download — `sandbox.fs.downloadFiles`.
const responses = await withProviderSpan({
name: "transfer",
wallMsAttr: SPAN_ATTR.transferWallMs,
attributes: {
[SPAN_ATTR.transferGuardCount]: guardRoundTrips,
[SPAN_ATTR.transferDirection]: "outbound",
},
run: () =>
sandbox.fs.downloadFiles([{ source: remoteTar, destination: localTar }], timeoutSeconds),
});
const response = responses.find((entry) => entry.source === remoteTar) ?? responses[0];
if (!response || response.error) {
throw new Error(

24
packages/plugins/sandbox-providers/daytona/src/manifest.ts
View File

@ -1,12 +1,16 @@
import type { PaperclipPluginManifestV1 } from "@paperclipai/plugin-sdk";
const PLUGIN_ID = "paperclip.daytona-sandbox-provider";
// 0.1.3 renames the login transport flag from `supportsSetupTokenLogin` to the
// neutral `supportsLoginPty`. The boot reconcile reads the persisted manifest
// raw and does not re-run the validator, so it never canonicalizes the old
// name. The version bump makes the reconcile refresh the persisted manifest for
// an existing install, so the renamed capability propagates.
const PLUGIN_VERSION = "0.1.3";
// The bundled-plugin boot reconcile refreshes the persisted manifest for an
// existing install only when PLUGIN_VERSION changes. A manifest change without a
// version bump never reaches an existing install. The reconcile also reads the
// persisted manifest raw and does not re-run the validator, so it never
// canonicalizes a renamed capability.
//
// 0.1.3 renamed the login transport flag from `supportsSetupTokenLogin` to the
// neutral `supportsLoginPty`.
// 0.1.4 adds the `concurrentSyncOperations` sandbox capability to the driver.
const PLUGIN_VERSION = "0.1.4";
const manifest: PaperclipPluginManifestV1 = {
id: PLUGIN_ID,
@ -33,8 +37,16 @@ const manifest: PaperclipPluginManifestV1 = {
// emits incremental session output while the command runs. Declare the
// opt-in capability so the host selects the session-output streaming path.
// A generic one-shot provider that omits this key keeps the poll path.
//
// Daytona also runs file transfers into and out of the sandbox in parallel.
// Each concurrent sync hook call uses separate temporary state (random
// scratch names and per-mapping host temporary directories), and teardown
// waits for all active calls. Declare the opt-in capability so the host may
// schedule sync operations concurrently. The host resolves it `true` only
// when the worker also verifies both sync verbs.
sandboxCapabilities: {
incrementalSessionOutput: true,
concurrentSyncOperations: true,
},
supportsInteractiveSetup: true,
interactiveSetupConnectionTypes: ["ssh"],

391
packages/plugins/sandbox-providers/daytona/src/plugin.test.ts
View File

@ -139,6 +139,15 @@ describe("Daytona sandbox provider plugin", () => {
});
});
it("declares the concurrent-sync-operations capability so the host may parallelize sync operations", () => {
// Daytona runs file transfers into and out of the sandbox in parallel, so it
// declares the opt-in capability. The host resolves it `true` only when the
// worker also verifies both sync verbs, which the sync hooks provide.
expect(manifest.environmentDrivers?.[0]?.sandboxCapabilities).toMatchObject({
concurrentSyncOperations: true,
});
});
it("normalizes config and validates the API key fallback", async () => {
process.env.DAYTONA_API_KEY = "host-key";
@ -3014,6 +3023,95 @@ describe("daytona native file-sync hooks", () => {
};
}
// A concurrency gate for a fake transfer call (uploadFiles / downloadFiles).
// The gate lets two concurrent hook calls both enter the fake, then holds them
// there until the test releases them. It records the peak number of calls that
// are in the fake at the same time, so a test proves the two calls overlap.
//
// `expected` is how many calls the test starts. `bothArrived` resolves once
// that many calls sit inside the fake at the same moment. `release()` frees
// them. `body` is the fake implementation: it marks arrival, waits for the
// release, and then runs `onRelease` to produce the fake result.
function createTransferGate<T>(expected: number, onRelease: (args: unknown[]) => Promise<T>) {
let inFlight = 0;
let peakInFlight = 0;
let signalArrived!: () => void;
const bothArrived = new Promise<void>((resolve) => {
signalArrived = resolve;
});
let signalReleased!: () => void;
const released = new Promise<void>((resolve) => {
signalReleased = resolve;
});
const body = async (...args: unknown[]): Promise<T> => {
inFlight += 1;
peakInFlight = Math.max(peakInFlight, inFlight);
if (inFlight === expected) signalArrived();
await released;
inFlight -= 1;
return onRelease(args);
};
return {
body,
bothArrived,
release: () => signalReleased(),
peak: () => peakInFlight,
};
}
// Write each download request's snapshot bytes to its host destination and
// report success, matching the real batch-download contract the outbound sync
// path expects.
async function fulfilDownload(args: unknown[]): Promise<Array<{ source: string; result: string }>> {
const requests = args[0] as Array<{ source: string; destination: string }>;
return Promise.all(
requests.map(async (request) => {
await fs.writeFile(request.destination, "bytes");
return { source: request.source, result: request.destination };
}),
);
}
function syncInParams(overrides: {
operationId: string;
sourcePath: string;
targetPath: string;
}) {
return {
driverKey: "daytona",
companyId: "company-1",
environmentId: "env-1",
config: { timeoutMs: 300000, reuseLease: false },
lease: syncLease(),
operations: [
{
operationId: overrides.operationId,
files: [{ sourcePath: overrides.sourcePath, targetPath: overrides.targetPath, kind: "file" as const }],
},
],
};
}
function syncOutParams(overrides: {
operationId: string;
sourcePath: string;
targetPath: string;
}) {
return {
driverKey: "daytona",
companyId: "company-1",
environmentId: "env-1",
config: { timeoutMs: 300000, reuseLease: false },
lease: syncLease(),
operations: [
{
operationId: overrides.operationId,
files: [{ sourcePath: overrides.sourcePath, targetPath: overrides.targetPath, kind: "file" as const }],
},
],
};
}
beforeEach(() => {
mockGet.mockReset();
process.env.DAYTONA_API_KEY = "host-key";
@ -3295,6 +3393,66 @@ describe("daytona native file-sync hooks", () => {
expect(spans.find((span) => span.name === "pack")).toBeUndefined();
});
it("marks the inbound transfer span with the inbound direction attribute", async () => {
const hostDir = await makeHostDir();
const source = path.join(hostDir, "config.txt");
await fs.writeFile(source, "plain");
const sandbox = createMockSandbox();
mockGet.mockResolvedValue(sandbox);
const { tracer, spans } = createRecordingPluginTracer();
const restore = __setDaytonaPluginContextForTest({ tracer } as unknown as PluginContext);
try {
await plugin.definition.onEnvironmentSyncIn?.(
syncInParams({
operationId: "sync-op-in-dir",
sourcePath: source,
targetPath: `${REMOTE_DIR}/config.txt`,
}),
);
} finally {
restore();
}
// An upload to the sandbox is an inbound transfer.
const transfer = spans.find((span) => span.name === "transfer");
expect(transfer).toBeDefined();
expect(transfer!.attributes["paperclip.sandbox.startup.transfer.direction"]).toBe("inbound");
});
it("marks the outbound transfer span with the outbound direction attribute", async () => {
const hostDir = await makeHostDir();
const sandbox = createMockSandbox();
sandbox.fs.downloadFiles.mockImplementation(async (requests: Array<{ source: string; destination?: string }>) => {
return Promise.all(
requests.map(async (req) => {
await fs.writeFile(req.destination!, "bytes");
return { source: req.source, result: req.destination };
}),
);
});
mockGet.mockResolvedValue(sandbox);
const { tracer, spans } = createRecordingPluginTracer();
const restore = __setDaytonaPluginContextForTest({ tracer } as unknown as PluginContext);
try {
await plugin.definition.onEnvironmentSyncOut?.(
syncOutParams({
operationId: "sync-op-out-dir",
sourcePath: `${REMOTE_DIR}/out/result.txt`,
targetPath: path.join(hostDir, "result.txt"),
}),
);
} finally {
restore();
}
// A download from the sandbox is an outbound transfer.
const transfer = spans.find((span) => span.name === "transfer");
expect(transfer).toBeDefined();
expect(transfer!.attributes["paperclip.sandbox.startup.transfer.direction"]).toBe("outbound");
});
it("opens a pack span and a transfer span around a directory mapping sync", async () => {
const hostDir = await makeHostDir();
const sourceDir = path.join(hostDir, "assets");
@ -4530,6 +4688,239 @@ describe("daytona native file-sync hooks", () => {
});
expect(withEmpty.process.executeCommand.mock.calls.length).toBe(baselineExecCount);
});
it("runs two concurrent inbound syncIn calls with separate reserved scratch names", async () => {
const hostDir = await makeHostDir();
const sourceA = path.join(hostDir, "a.txt");
const sourceB = path.join(hostDir, "b.txt");
await fs.writeFile(sourceA, "alpha");
await fs.writeFile(sourceB, "beta");
const sandbox = createMockSandbox();
// Gate the upload so both concurrent calls sit inside uploadFiles together.
const gate = createTransferGate(2, async () => undefined);
sandbox.fs.uploadFiles.mockImplementation(gate.body);
mockGet.mockResolvedValue(sandbox);
const callA = plugin.definition.onEnvironmentSyncIn?.(
syncInParams({ operationId: "in-a", sourcePath: sourceA, targetPath: `${REMOTE_DIR}/a.txt` }),
);
const callB = plugin.definition.onEnvironmentSyncIn?.(
syncInParams({ operationId: "in-b", sourcePath: sourceB, targetPath: `${REMOTE_DIR}/b.txt` }),
);
// Both calls reached the upload before either finished, so they overlap.
await gate.bothArrived;
expect(gate.peak()).toBe(2);
expect(sandbox.fs.uploadFiles).toHaveBeenCalledTimes(2);
gate.release();
await Promise.all([callA, callB]);
// Each concurrent call staged its upload under its own reserved scratch name;
// the two calls never share a temporary destination.
const destinations = sandbox.fs.uploadFiles.mock.calls.flatMap(
([uploads]) => (uploads as Array<{ destination: string }>).map((upload) => upload.destination),
);
expect(destinations).toHaveLength(2);
for (const destination of destinations) {
expect(path.posix.basename(destination)).toMatch(/^\.paperclip-upload-/);
expect(path.posix.dirname(destination)).toBe(REMOTE_DIR);
}
expect(new Set(destinations).size).toBe(destinations.length);
});
it("runs two concurrent outbound syncOut calls that both reach downloadFiles before either opens", async () => {
const hostDir = await makeHostDir();
const targetA = path.join(hostDir, "a.txt");
const targetB = path.join(hostDir, "b.txt");
const sandbox = createMockSandbox();
// Gate the download so both concurrent calls sit inside downloadFiles
// together before either resolves.
const gate = createTransferGate(2, fulfilDownload);
sandbox.fs.downloadFiles.mockImplementation(gate.body);
mockGet.mockResolvedValue(sandbox);
const callA = plugin.definition.onEnvironmentSyncOut?.(
syncOutParams({ operationId: "out-a", sourcePath: `${REMOTE_DIR}/a.txt`, targetPath: targetA }),
);
const callB = plugin.definition.onEnvironmentSyncOut?.(
syncOutParams({ operationId: "out-b", sourcePath: `${REMOTE_DIR}/b.txt`, targetPath: targetB }),
);
// Both calls reached the download before either gate opened, so they overlap.
await gate.bothArrived;
expect(gate.peak()).toBe(2);
expect(sandbox.fs.downloadFiles).toHaveBeenCalledTimes(2);
gate.release();
await Promise.all([callA, callB]);
// Each concurrent call read its own reserved snapshot; the two calls never
// share a download source.
const sources = sandbox.fs.downloadFiles.mock.calls.flatMap(
([requests]) => (requests as Array<{ source: string }>).map((request) => request.source),
);
expect(sources).toHaveLength(2);
for (const source of sources) {
expect(source.startsWith(`${REMOTE_DIR}/`)).toBe(true);
expect(path.posix.basename(source)).toMatch(/^\.paperclip-upload-/);
}
expect(new Set(sources).size).toBe(sources.length);
expect(await fs.readFile(targetA, "utf8")).toBe("bytes");
expect(await fs.readFile(targetB, "utf8")).toBe("bytes");
});
it("waits for one active inbound and one active outbound call before teardown releases the sandbox", async () => {
const hostDir = await makeHostDir();
const inboundSource = path.join(hostDir, "in.txt");
const outboundTarget = path.join(hostDir, "out.txt");
await fs.writeFile(inboundSource, "inbound");
const sandbox = createMockSandbox({ id: "sandbox-123" });
// Hold the inbound upload and the outbound download open at the same time, so
// the shared lease has two active sync calls when teardown starts.
let releaseUpload!: () => void;
sandbox.fs.uploadFiles.mockImplementation(async () => {
await new Promise<void>((resolve) => {
releaseUpload = resolve;
});
});
let releaseDownload!: () => void;
sandbox.fs.downloadFiles.mockImplementation(async (requests: Array<{ source: string; destination: string }>) => {
await new Promise<void>((resolve) => {
releaseDownload = resolve;
});
return Promise.all(
requests.map(async (request) => {
await fs.writeFile(request.destination, "bytes");
return { source: request.source, result: request.destination };
}),
);
});
mockGet.mockResolvedValue(sandbox);
const inboundCall = plugin.definition.onEnvironmentSyncIn?.(
syncInParams({ operationId: "in-active", sourcePath: inboundSource, targetPath: `${REMOTE_DIR}/in.txt` }),
);
const outboundCall = plugin.definition.onEnvironmentSyncOut?.(
syncOutParams({ operationId: "out-active", sourcePath: `${REMOTE_DIR}/out.txt`, targetPath: outboundTarget }),
);
// Let both sync calls register on the activity gate and reach their hung
// transfer, so teardown sees a refCount of two.
await new Promise((resolve) => setTimeout(resolve, 0));
const destroyCall = plugin.definition.onEnvironmentDestroyLease?.({
driverKey: "daytona",
companyId: "company-1",
environmentId: "env-1",
providerLeaseId: "sandbox-123",
config: { timeoutMs: 300000, reuseLease: false },
});
await new Promise((resolve) => setTimeout(resolve, 0));
// Two active sync calls block teardown, so it must not delete the sandbox yet.
expect(sandbox.delete).not.toHaveBeenCalled();
// Release only the inbound call. One outbound call is still active, so
// teardown must keep waiting.
releaseUpload();
await inboundCall;
await new Promise((resolve) => setTimeout(resolve, 0));
expect(sandbox.delete).not.toHaveBeenCalled();
// Release the outbound call. No sync call is active now, so teardown deletes.
releaseDownload();
await Promise.all([outboundCall, destroyCall]);
expect(sandbox.fs.uploadFiles).toHaveBeenCalledTimes(1);
expect(sandbox.fs.downloadFiles).toHaveBeenCalledTimes(1);
expect(sandbox.delete).toHaveBeenCalledTimes(1);
});
it("opens a transfer span with the guard round-trip count around the bulk file download", async () => {
const hostDir = await makeHostDir();
const target = path.join(hostDir, "result.txt");
const sandbox = createMockSandbox();
sandbox.fs.downloadFiles.mockImplementation(async (requests: Array<{ source: string; destination: string }>) => {
return Promise.all(
requests.map(async (request) => {
await fs.writeFile(request.destination, "bytes");
return { source: request.source, result: request.destination };
}),
);
});
mockGet.mockResolvedValue(sandbox);
const { tracer, spans } = createRecordingPluginTracer();
const restore = __setDaytonaPluginContextForTest({ tracer } as unknown as PluginContext);
try {
await plugin.definition.onEnvironmentSyncOut?.({
driverKey: "daytona",
companyId: "company-1",
environmentId: "env-1",
config: { timeoutMs: 300000, reuseLease: false },
lease: syncLease(),
operations: [
{
operationId: "sync-op-out",
files: [{ sourcePath: `${REMOTE_DIR}/out/result.txt`, targetPath: target, kind: "file" }],
},
],
});
} finally {
restore();
}
const transfer = spans.find((span) => span.name === "transfer");
expect(transfer).toBeDefined();
expect(transfer!.ended).toBe(true);
// One serial guard round trip before the transfer: the validate-and-snapshot.
expect(transfer!.attributes["paperclip.sandbox.startup.transfer.guard.count"]).toBe(1);
expect(transfer!.attributes["paperclip.sandbox.startup.provider"]).toBe("daytona");
expect(typeof transfer!.attributes["paperclip.sandbox.startup.transfer.wall_ms"]).toBe("number");
});
it("opens a transfer span around a directory-mapping download with the guard round-trip count", async () => {
const hostDir = await makeHostDir();
const targetDir = path.join(hostDir, "assets");
const sandbox = createMockSandbox();
sandbox.fs.downloadFiles.mockImplementation(async (requests: Array<{ source: string; destination: string }>) => {
// Write a valid empty tar (1024-byte zero EOF marker) so host-side extract
// is a clean no-op.
await Promise.all(requests.map((request) => fs.writeFile(request.destination, Buffer.alloc(1024))));
return requests.map((request) => ({ source: request.source, result: request.destination }));
});
mockGet.mockResolvedValue(sandbox);
const { tracer, spans } = createRecordingPluginTracer();
const restore = __setDaytonaPluginContextForTest({ tracer } as unknown as PluginContext);
try {
await plugin.definition.onEnvironmentSyncOut?.({
driverKey: "daytona",
companyId: "company-1",
environmentId: "env-1",
config: { timeoutMs: 300000, reuseLease: false },
lease: syncLease(),
operations: [
{
operationId: "sync-op-out-dir",
files: [{ sourcePath: `${REMOTE_DIR}/out/assets`, targetPath: targetDir, kind: "directory" }],
},
],
});
} finally {
restore();
}
const transfer = spans.find((span) => span.name === "transfer");
expect(transfer).toBeDefined();
expect(transfer!.ended).toBe(true);
// Two serial guard round trips before the transfer: confinement + in-sandbox
// tar.
expect(transfer!.attributes["paperclip.sandbox.startup.transfer.guard.count"]).toBe(2);
expect(typeof transfer!.attributes["paperclip.sandbox.startup.transfer.wall_ms"]).toBe("number");
});
});
describe("daytona manifest memory config", () => {

41
packages/shared/src/telemetry/README.md
View File

@ -111,10 +111,16 @@ absent, never a misleading `0`.
| `stage.sync` | Workspace stage-sync step. | `sandbox.startup` |
| `snapshot.git` | Host-side git workspace enumeration inside `stage.sync` (`git status --ignored`, the HEAD diffs, `ls-files`). | `stage.sync` |
| `snapshot.baseline` | Host-side baseline workspace content-hash walk inside `stage.sync`, kept for restore. | `stage.sync` |
| `pack` | Host-side workspace tarball build inside `stage.sync`. | `stage.sync` |
| `stage.workspace` | One inbound workspace stage task inside `stage.sync`. It packs and uploads the workspace. | `stage.sync` |
| `stage.asset.<key>` | One inbound asset stage task inside `stage.sync`. It packs and uploads one managed-home asset. The `<key>` segment is the asset key. | `stage.sync` |
| `stage.project.<id>` | One inbound referenced-project stage task inside `stage.sync`. It uploads one referenced project. The `<id>` segment is the project id. | `stage.sync` |
| `pack` | Host-side workspace tarball build inside the `stage.workspace` task. | `stage.workspace` |
| `bridge.paperclip` | Paperclip bridge start step. | `sandbox.startup` |
| `bridge.process-session` | Process-session bridge start step. | `sandbox.startup` |
| `acp.handshake` | ACP session handshake step. | `sandbox.startup` |
| `sandbox.syncBack` | The settlement sync-back that restores the managed home at teardown. | the active run span |
| `restore.workspace` | One outbound workspace restore task at teardown. It reads the sandbox workspace back and merges it into the host workspace. | `sandbox.syncBack` |
| `restore.asset.<key>` | One outbound asset restore task at teardown. It reads one asset back to its host store. The `<key>` segment is the asset key. | `sandbox.syncBack` |
| `sandbox.agentSession.sendInput` | One outbound ACP message to the agent — the socket handler's one `writeTextFile` exec. | the active run span |
| `sandbox.agentSession.pollOutput` | One 100 ms poll tick — `list`, then `read`+`remove` per file found (`1 + 2n` execs). | the active run span |
| `sandbox.callbackBridge.relayRequest` | One Paperclip-API callback request — read the request, write the response, remove it. | the active run span |
@ -123,9 +129,19 @@ absent, never a misleading `0`.
A step span name is the step name. The `sandbox.exec` span parents to the step
span that runs the execution, so each execution nests under its step. Within
`stage.sync`, the host-side sub-steps `snapshot.git`, `snapshot.baseline`, and
`pack` open as child spans of the step, so the host work at the head of the step
is attributed rather than showing as a gap. A run-time
`stage.sync`, the host-side sub-steps `snapshot.git` and `snapshot.baseline` open
as child spans of the step, so the host work at the head of the step is
attributed rather than showing as a gap. Each inbound sync operation also opens
its own task span under `stage.sync`: `stage.workspace`, one `stage.asset.<key>`
per asset, and one `stage.project.<id>` per referenced project. The `pack` span
nests under `stage.workspace`, because the host builds the tarball inside that
task. Two concurrent tasks produce overlapping spans.
The settlement `sandbox.syncBack` span runs at teardown and parents to the run
span. It wraps the managed-home restore. Each outbound restore operation opens
its own task span under `sandbox.syncBack`: `restore.workspace` and one
`restore.asset.<key>` per asset. Two concurrent restore tasks produce overlapping
spans. A run-time
`sandbox.exec` span parents instead to the run-time wrapper span that runs it
(`sandbox.agentSession.sendInput`, `sandbox.agentSession.pollOutput`,
`sandbox.callbackBridge.relayRequest`, or `sandbox.agentProcess`). Each run-time
@ -228,7 +244,7 @@ before it records the span.
| Span | Scope | Parent |
| --- | --- | --- |
| `sandbox.daytona.pack` | The host-local pack step that builds the upload tarball. It makes no sandbox round trip. | the active startup step span |
| `sandbox.daytona.transfer` | The transfer step that uploads the files to the sandbox. | the active startup step span |
| `sandbox.daytona.transfer` | The transfer step: an upload to the sandbox (inbound) or a download from the sandbox (outbound). The `paperclip.sandbox.startup.transfer.direction` attribute records the direction. | the active sync task span (`stage.*` inbound, `restore.*` under `sandbox.syncBack` outbound) |
| `sandbox.daytona.ensureDirectory` | The `mkdir -p` step that ensures a directory exists before a write. | the active startup step span |
| `sandbox.daytona.checkSymlinkEscape` | The re-check step that a path resolves inside the workspace root before use. | the active startup step span |
| `sandbox.daytona.promote` | The atomic move of a staged temp onto its target via a pinned dir handle. | the active startup step span |
@ -258,10 +274,12 @@ the attributes that the producer sends for one span.
| `paperclip.sandbox.startup.pack.wall_ms` | number | yes | The host-local wall time of the pack step. It rides the `sandbox.daytona.pack` span. |
| `paperclip.sandbox.startup.transfer.wall_ms` | number | yes | The wall time of the transfer step. It rides the `sandbox.daytona.transfer` span. |
| `paperclip.sandbox.startup.transfer.guard.count` | number | yes | The number of serial guard round trips before one transfer. It rides the `sandbox.daytona.transfer` span. |
| `paperclip.sandbox.startup.transfer.direction` | string | yes | The transfer direction (`inbound` or `outbound`). It rides the `sandbox.daytona.transfer` span. |
The `span.record` host handler enforces the allowlist. It re-maps `provider`
through the provider-family normalizer. It keeps `outcome` only when the value
is `ok`, `skipped`, or `failed`. It keeps a numeric attribute only when the
is `ok`, `skipped`, or `failed`. It keeps `transfer.direction` only when the
value is `inbound` or `outbound`. It keeps a numeric attribute only when the
value is a finite number. It drops a status message and keeps only the numeric
status code. The handler never throws, because observability must not change the
sync control flow.
@ -271,9 +289,14 @@ capability. So only a plugin that registers an environment driver may emit a
provider span. The capability gate rejects a provider span from any other
plugin.
The host parents each provider span to the active startup step span. The host
mints a W3C `traceparent` from the active step and passes it to the plugin
worker on the per-call invocation channel. The worker tags its span with the
The host parents each provider span to the active sync task span. An inbound
transfer runs inside a `stage.*` task span, so its provider spans parent there.
An outbound transfer runs inside a `restore.*` task span under `sandbox.syncBack`
at teardown, so its provider spans parent there. The host mints a W3C
`traceparent` from the active task span and passes it to the plugin worker on the
per-call invocation channel. The teardown restore runs inside the run-parented
`sandbox.syncBack` span, so the host mints a `traceparent` for an outbound
provider span the same way it does for an inbound one. The worker tags its span with the
`traceparent` and treats the value as opaque. The worker never derives the
parent from it. The host recovers the `traceparent` from its own invocation
record, so a worker can never forge a parent. The host validates the

10
packages/shared/src/types/plugin.ts
View File

@ -163,6 +163,16 @@ export interface SandboxProviderCapabilities {
* the output-file poll path.
*/
incrementalSessionOutput?: boolean;
/**
* Provider can run file transfers into and out of the sandbox in parallel, in
* both directions. This is an opt-in behavioral guarantee. An omitted key
* denies the capability, so the host keeps the serial transfer path. The host
* resolves the capability `true` only when the provider declares this key
* `true` and the live worker verifies both sync verbs (`environmentSyncIn` and
* `environmentSyncOut`). A provider that verifies only one verb resolves
* `false`.
*/
concurrentSyncOperations?: boolean;
}
export interface PluginEnvironmentDriverDeclaration {

35
packages/shared/src/validators/plugin.test.ts
View File

@ -313,18 +313,29 @@ describe("sandbox provider capability declaration validators", () => {
expect(rejected.success).toBe(false);
});
it("test_removed_concurrency_capabilities_are_rejected_as_unknown_keys", () => {
// The concurrency flags left the public contract because no runtime path
// enforced them. The strict schema now rejects them, so a manifest cannot
// declare a capability the host does not honor.
for (const key of ["concurrentSyncAndExec", "concurrentSyncOperations"]) {
const rejected = pluginManifestV1Schema.safeParse(
buildSandboxProviderManifest({
sandboxCapabilities: { [key]: true },
}),
);
expect(rejected.success).toBe(false);
}
it("test_manifest_accepts_concurrent_sync_operations_capability", () => {
// A provider opts in to parallel bidirectional file sync with this key. The
// strict schema accepts it and keeps the declared value.
const parsed = pluginManifestV1Schema.parse(
buildSandboxProviderManifest({
sandboxCapabilities: { concurrentSyncOperations: true },
}),
);
expect(parsed.environmentDrivers?.[0]?.sandboxCapabilities).toEqual({
concurrentSyncOperations: true,
});
});
it("test_manifest_rejects_unknown_sync_concurrency_capability_key", () => {
// A neighboring but unknown concurrency key must fail validation, not drop
// silently. The strict schema rejects a capability the host does not honor.
const rejected = pluginManifestV1Schema.safeParse(
buildSandboxProviderManifest({
sandboxCapabilities: { concurrentSyncAndExec: true },
}),
);
expect(rejected.success).toBe(false);
});
it("test_supports_reusable_leases_compat_maps_to_reusable_leases", () => {

1
packages/shared/src/validators/plugin.ts
View File

@ -166,6 +166,7 @@ export const sandboxProviderCapabilitiesSchema = z.object({
persistentProcessSessions: z.boolean().optional(),
independentControlCommands: z.boolean().optional(),
incrementalSessionOutput: z.boolean().optional(),
concurrentSyncOperations: z.boolean().optional(),
}).strict();
export type SandboxProviderCapabilitiesInput = z.infer<typeof sandboxProviderCapabilitiesSchema>;

40
server/src/__tests__/environment-execution-target-capabilities.test.ts
View File

@ -19,6 +19,9 @@ const SNAPSHOT: EffectiveSandboxCapabilities = {
persistentProcessSessions: true,
independentControlCommands: false,
incrementalSessionOutput: false,
// Concurrent sync operations need BOTH sync verbs; this snapshot verified only
// inbound sync, so the opt-in stays off.
concurrentSyncOperations: false,
};
// A snapshot that grants every capability. A test overrides one flag to prove
@ -30,6 +33,7 @@ const FULL_GRANT: EffectiveSandboxCapabilities = {
persistentProcessSessions: true,
independentControlCommands: true,
incrementalSessionOutput: true,
concurrentSyncOperations: true,
};
// Build a sandbox execution target with a fixed snapshot and a fixed
@ -185,6 +189,42 @@ describe("effective snapshot gates the sync decision", () => {
});
});
// The execution target carries the concurrent-sync opt-in on both the effective
// snapshot and the runner. The runner Boolean feeds the sync client, which then
// tells the orchestrator whether it may run sync operations concurrently.
describe("effective snapshot carries the concurrent-sync capability", () => {
beforeEach(() => {
mockResolveEnvironmentDriverConfigForRuntime.mockReset();
});
it("carries the concurrent-sync opt-in on the snapshot and the runner", async () => {
const { target } = await buildSandboxTarget({ snapshot: FULL_GRANT, supportsSync: true });
expect(target.effectiveCapabilities?.concurrentSyncOperations).toBe(true);
expect(target.runner?.allowConcurrentSyncOperations).toBe(true);
});
it("keeps the concurrent-sync opt-in off when the snapshot removes it", async () => {
const { target } = await buildSandboxTarget({
snapshot: { ...FULL_GRANT, concurrentSyncOperations: false },
supportsSync: true,
});
expect(target.effectiveCapabilities?.concurrentSyncOperations).toBe(false);
expect(target.runner?.allowConcurrentSyncOperations).toBe(false);
});
it("keeps the concurrent-sync opt-in off when the resolution rejects", async () => {
// A rejected resolution carries no snapshot; it must not read as an open
// grant. The runner keeps the opt-in off.
const { target } = await buildSandboxTarget({
snapshot: null,
supportsSync: true,
rejectResolution: true,
});
expect(target.effectiveCapabilities).toBeUndefined();
expect(target.runner?.allowConcurrentSyncOperations).toBe(false);
});
});
// Session-output streaming is decided downstream from the carried snapshot
// alone (see `streamAgentSessionOutput` in `acpx-engine/execute.ts`): the bridge
// streams only when the snapshot grants `incrementalSessionOutput`. That key is

5
server/src/__tests__/heartbeat-project-env.test.ts
View File

@ -1019,17 +1019,18 @@ describe("buildReferencedProjectRunObservability", () => {
failures: [
{ projectId: "project-b", reason: "authorization" },
{ projectId: "project-c", reason: "resolution" },
{ projectId: "project-d", reason: "staging" },
{ projectId: "project-d", reason: "staging", error: "extract failed: boom" },
],
});
// Requested is the synced count plus every dropped project, so the counts reconcile.
expect(observability.referenced_projects_requested).toBe(4);
expect(observability.referenced_projects_synced).toBe(1);
// A staging failure carries its error message; a failure without one omits the field.
expect(observability.referenced_project_failures).toEqual([
{ project_id: "project-b", reason: "authorization" },
{ project_id: "project-c", reason: "resolution" },
{ project_id: "project-d", reason: "staging" },
{ project_id: "project-d", reason: "staging", error: "extract failed: boom" },
]);
});

17
server/src/__tests__/plugin-host-services-span.test.ts
View File

@ -311,4 +311,21 @@ describe("clampProviderSpanAttributes", () => {
// A non-finite number yields no attribute; `exec.command` is not allowed.
});
});
it("keeps the transfer direction for each closed-set value", () => {
expect(clampProviderSpanAttributes({ [A.transferDirection]: "inbound" })).toEqual({
[A.transferDirection]: "inbound",
});
expect(clampProviderSpanAttributes({ [A.transferDirection]: "outbound" })).toEqual({
[A.transferDirection]: "outbound",
});
});
it("drops a transfer direction outside the closed set or of the wrong type", () => {
// A free-form string, an empty string, and a non-string all yield no
// attribute, so the direction stays bounded and low-cardinality.
expect(clampProviderSpanAttributes({ [A.transferDirection]: "sideways" })).toEqual({});
expect(clampProviderSpanAttributes({ [A.transferDirection]: "" })).toEqual({});
expect(clampProviderSpanAttributes({ [A.transferDirection]: 1 })).toEqual({});
});
});

35
server/src/__tests__/sandbox-capability-contract.test.ts
View File

@ -290,6 +290,40 @@ describe("sandbox capability contract normalizer", () => {
expect(effective.incrementalSessionOutput).toBe(false);
});
it("test_concurrent_sync_operations_is_opt_in_and_needs_both_sync_verbs", () => {
// Parallel bidirectional file sync is opt-in and direction-neutral. It needs
// both sync verbs, so a provider that verifies only one direction cannot get
// the capability. An absent declaration denies it even with both verbs.
const undeclared = resolveEffectiveSandboxCapabilities({
verifiedMethods: ["environmentSyncIn", "environmentSyncOut"],
declared: null,
});
expect(undeclared.concurrentSyncOperations).toBe(false);
// A positive declaration with both verified verbs resolves true.
const bothVerbs = resolveEffectiveSandboxCapabilities({
verifiedMethods: ["environmentSyncIn", "environmentSyncOut"],
declared: { concurrentSyncOperations: true },
});
expect(bothVerbs.concurrentSyncOperations).toBe(true);
// Only the inbound verb: the outbound prerequisite is missing, so it resolves
// false.
const inOnly = resolveEffectiveSandboxCapabilities({
verifiedMethods: ["environmentSyncIn"],
declared: { concurrentSyncOperations: true },
});
expect(inOnly.concurrentSyncOperations).toBe(false);
// Only the outbound verb: the inbound prerequisite is missing, so it resolves
// false.
const outOnly = resolveEffectiveSandboxCapabilities({
verifiedMethods: ["environmentSyncOut"],
declared: { concurrentSyncOperations: true },
});
expect(outOnly.concurrentSyncOperations).toBe(false);
});
it("test_unknown_or_unavailable_verification_resolves_false", () => {
const declaredAll = {
reusableLeases: true,
@ -298,6 +332,7 @@ describe("sandbox capability contract normalizer", () => {
persistentProcessSessions: true,
independentControlCommands: true,
incrementalSessionOutput: true,
concurrentSyncOperations: true,
};
for (const verifiedMethods of [null, undefined, [] as string[]]) {

6
server/src/services/environment-execution-target.ts
View File

@ -317,6 +317,12 @@ export async function resolveEnvironmentExecutionTarget(input: {
// here. The client falls back to the chunked upload path when this is
// false.
supportsSingleStreamStdinProgress: false,
// Carry the verified concurrent-sync opt-in to the sync client. The
// client copies it onto the native path and ignores it on the base64
// fallback, which always permits concurrency. A null snapshot or a
// provider that never opted in keeps it false, so an unverified
// provider never permits concurrent sync operations.
allowConcurrentSyncOperations: effectiveCapabilities?.concurrentSyncOperations === true,
execute: async (commandInput) => {
// Record true start and stop timestamps around the provider await,
// so the exec span and the result carry a real wall time.

10
server/src/services/environment-runtime.ts
View File

@ -89,6 +89,7 @@ export const SANDBOX_CAPABILITY_KEYS = [
"persistentProcessSessions",
"independentControlCommands",
"incrementalSessionOutput",
"concurrentSyncOperations",
] as const;
export type SandboxCapabilityKey = (typeof SANDBOX_CAPABILITY_KEYS)[number];
@ -104,6 +105,7 @@ export type SandboxCapabilityKey = (typeof SANDBOX_CAPABILITY_KEYS)[number];
*/
const SANDBOX_CAPABILITY_OPT_IN_KEYS: ReadonlySet<SandboxCapabilityKey> = new Set([
"incrementalSessionOutput",
"concurrentSyncOperations",
]);
/**
@ -134,6 +136,12 @@ const SANDBOX_CAPABILITY_OPT_IN_KEYS: ReadonlySet<SandboxCapabilityKey> = new Se
* provider tails the session log through it. The verified verb is necessary
* but not sufficient: this key is opt-in, so the declaration is the real gate
* (see {@link SANDBOX_CAPABILITY_OPT_IN_KEYS}).
* - `concurrentSyncOperations` requires BOTH sync verbs, because parallel
* bidirectional transfer runs an inbound and an outbound transfer at the same
* time. The two verbs are separate required groups, so a provider that
* verifies only one direction cannot get the capability. The verbs are
* necessary but not sufficient: this key is opt-in, so the declaration is the
* real gate (see {@link SANDBOX_CAPABILITY_OPT_IN_KEYS}).
*/
const SANDBOX_CAPABILITY_PREREQUISITE_METHODS: Record<SandboxCapabilityKey, readonly (readonly string[])[]> = {
// Reusable leases require ALL reuse verbs. Each verb is its own required
@ -147,6 +155,7 @@ const SANDBOX_CAPABILITY_PREREQUISITE_METHODS: Record<SandboxCapabilityKey, read
persistentProcessSessions: [["environmentExecute"]],
independentControlCommands: [["environmentExecute"]],
incrementalSessionOutput: [["environmentExecute"]],
concurrentSyncOperations: [["environmentSyncIn"], ["environmentSyncOut"]],
};
function capabilityIsVerified(
@ -230,6 +239,7 @@ export function resolveEffectiveSandboxCapabilities(input: {
persistentProcessSessions: resolve("persistentProcessSessions"),
independentControlCommands: resolve("independentControlCommands"),
incrementalSessionOutput: resolve("incrementalSessionOutput"),
concurrentSyncOperations: resolve("concurrentSyncOperations"),
};
}

12
server/src/services/heartbeat.ts
View File

@ -2865,6 +2865,12 @@ export type ReferencedProjectFailureReason = "authorization" | "resolution" | "s
export interface ReferencedProjectFailure {
projectId: string;
reason: ReferencedProjectFailureReason;
/**
* The failure message, when the layer that dropped the project produced one. A `staging` failure
* carries the remote extract or sync error here, so a reader of the run log learns why the project
* dropped. An `authorization` or `resolution` drop omits this field.
*/
error?: string;
}
export interface ResolvedRunReferencedProjects {
@ -3209,6 +3215,8 @@ export interface ReferencedProjectRunObservability {
referenced_project_failures: Array<{
project_id: string;
reason: ReferencedProjectFailureReason;
/** The failure message for a `staging` drop; absent for an `authorization` or `resolution` drop. */
error?: string;
}>;
}
@ -3231,6 +3239,9 @@ export function buildReferencedProjectRunObservability(input: {
referenced_project_failures: input.failures.map((failure) => ({
project_id: failure.projectId,
reason: failure.reason,
// Carry the error only when the layer produced one, so an authorization or resolution drop
// stays a two-field entry and a staging drop names its reason.
...(failure.error !== undefined ? { error: failure.error } : {}),
})),
};
}
@ -16067,6 +16078,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
failures: referencedProjectStagingFailures.map((failure) => ({
projectId: failure.projectId,
reason: "staging" as const,
error: failure.error,
})),
});
logger.info(

13
server/src/services/plugin-host-services.ts
View File

@ -545,6 +545,7 @@ const PROVIDER_SPAN_ATTR_ALLOWLIST: ReadonlySet<string> = new Set<string>([
SPAN_ATTRS.packWallMs,
SPAN_ATTRS.transferWallMs,
SPAN_ATTRS.transferGuardCount,
SPAN_ATTRS.transferDirection,
]);
/** The subset of allowed keys that carry a finite number. */
@ -557,11 +558,15 @@ const PROVIDER_SPAN_NUMERIC_ATTRS: ReadonlySet<string> = new Set<string>([
/** The closed value set for the `outcome` attribute. */
const KNOWN_SPAN_OUTCOMES: ReadonlySet<string> = new Set(["ok", "skipped", "failed"]);
/** The closed value set for the `transfer.direction` attribute. */
const KNOWN_TRANSFER_DIRECTIONS: ReadonlySet<string> = new Set(["inbound", "outbound"]);
/**
* Re-clamp the worker-sent attributes at the trust boundary. Drop every key that
* is not on the allowlist. Re-map `provider` through `normalizeProviderFamily`,
* bound `outcome` to its closed set, and keep a numeric attribute only when it
* is a finite number. The result holds only bounded, low-cardinality values.
* bound `outcome` and `transfer.direction` each to its closed set, and keep a
* numeric attribute only when it is a finite number. The result holds only
* bounded, low-cardinality values.
*/
export function clampProviderSpanAttributes(
raw: Record<string, unknown> | undefined,
@ -578,6 +583,10 @@ export function clampProviderSpanAttributes(
if (typeof value === "string" && KNOWN_SPAN_OUTCOMES.has(value)) clamped[key] = value;
continue;
}
if (key === SPAN_ATTRS.transferDirection) {
if (typeof value === "string" && KNOWN_TRANSFER_DIRECTIONS.has(value)) clamped[key] = value;
continue;
}
if (PROVIDER_SPAN_NUMERIC_ATTRS.has(key)) {
if (typeof value === "number" && Number.isFinite(value)) clamped[key] = value;
continue;