diff --git a/packages/adapter-utils/src/acpx-engine/execute.test.ts b/packages/adapter-utils/src/acpx-engine/execute.test.ts index f08288b3d0..c1fe7c2c2d 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.test.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.test.ts @@ -1851,6 +1851,85 @@ describe("ACPX engine remote sandbox staging seam (PR 1: workspace + cwd)", () = expect(sessionInputs[0]?.cwd).toBe(remoteCwd); }); + it("hands the merged paperclip env to the process-session launch when the setups overlap", async () => { + const { stateDir, localCwd, remoteCwd, executionTarget } = await setupRemoteSandbox(); + // Decode the process-session LAUNCH payload (the base64 command blob) — the + // in-sandbox process env is carried there, NOT in the exec's own `env`. + let launchPayload: Record | null = null; + (executionTarget as { runner: unknown }).runner = createLocalSandboxRunner((input) => { + if (input.env?.PAPERCLIP_SANDBOX_EXEC_CHANNEL === "bridge") { + const script = input.args?.[1] ?? ""; + const match = script.match(/PAPERCLIP_PROCESS_SESSION_COMMAND_B64='([^']+)'/); + if (match) { + launchPayload = JSON.parse(Buffer.from(match[1]!, "base64").toString("utf8")) as Record< + string, + unknown + >; + } + } + }); + + await runExecutor( + { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir, cwd: localCwd }, + { authToken: "real-run-jwt", executionTarget }, + ); + + // The process-session bridge receives its launch env as a DEFERRED thunk — + // the seam that lets its env-independent setup overlap the paperclip bridge + // start instead of running strictly after it. + const processArgs = vi.mocked(startAdapterExecutionTargetProcessSessionBridge).mock.calls[0]![0]; + expect(typeof processArgs.env).toBe("function"); + + // ...and despite the overlap the launch still observes the MERGED paperclip + // env: the paperclip-`env` → process-session-launch hand-off stays sequenced + // under concurrency (bridge base URL + minted bridge token both present, and + // the token is NOT the host run JWT). + const payloadEnv = ((launchPayload as Record | null)?.env ?? {}) as Record< + string, + unknown + >; + expect(payloadEnv).toMatchObject({ PAPERCLIP_API_BRIDGE_MODE: "queue_v1" }); + expect(String(payloadEnv.PAPERCLIP_API_URL ?? "")).toMatch(/^http:\/\/127\.0\.0\.1:\d+$/); + expect(payloadEnv.PAPERCLIP_API_KEY).toBeTruthy(); + expect(payloadEnv.PAPERCLIP_API_KEY).not.toBe("real-run-jwt"); + }); + + it("stops the process-session bridge when the paperclip bridge fails under concurrency", async () => { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + // The paperclip bridge fails; the process-session bridge — started CONCURRENTLY + // with it — still resolves a live handle. The abandon path must stop that + // handle so no started bridge leaks on partial failure. + const stop = vi.fn(async () => {}); + vi.mocked(startAdapterExecutionTargetPaperclipBridge).mockImplementationOnce(async () => { + throw new Error("paperclip bridge boom"); + }); + vi.mocked(startAdapterExecutionTargetProcessSessionBridge).mockImplementationOnce( + async () => ({ agentCommand: null, stop }) as never, + ); + + const execute = createAcpxEngineExecutor({ + createRuntime: () => buildRuntime() as never, + }); + + await expect( + execute({ + runId: "run-bridge-fail", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir, cwd: localCwd }, + context: {}, + authToken: "real-run-jwt", + executionTarget, + onLog: async () => {}, + onMeta: async () => {}, + onEvent: async () => {}, + } as never), + ).rejects.toThrow("paperclip bridge boom"); + + // The concurrently-started process-session bridge was stopped exactly once. + expect(stop).toHaveBeenCalledTimes(1); + }); + it("test_remote_session_new_uses_in_sandbox_cwd", async () => { const { stateDir, localCwd, remoteCwd, executionTarget } = await setupRemoteSandbox(); const { sessionInputs, runtimeOptions } = await runExecutor( @@ -2795,7 +2874,7 @@ describe("ACPX engine per-step startup timing (run.startup.step events)", () => } }); - it("carries per-step roundTrips + provider durations sourced from the sandbox runner counter", async () => { + it("carries roundTrips + provider durations for sequential startup steps and keeps concurrent bridge steps duration-only", async () => { const root = await makeTempRoot(); const stateDir = path.join(root, "state"); const localCwd = path.join(root, "worktree"); @@ -2818,10 +2897,13 @@ describe("ACPX engine per-step startup timing (run.startup.step events)", () => const steps = stepEvents(events); const seen = new Map(steps.map((event) => [String(event.payload?.step), event])); - // Every timed boundary carries a numeric roundTrips (the runner exposes - // execCount), even the ones that never exec. + // Every timed boundary still records duration. for (const event of steps) { - expect(typeof event.payload?.roundTrips).toBe("number"); + expect(typeof event.payload?.durationMs).toBe("number"); + } + // Sequential boundaries retain runner-counter attribution. + for (const step of ["workspace.resolve", "stage.sync", "acp.handshake"]) { + expect(typeof seen.get(step)?.payload?.roundTrips).toBe("number"); } // workspace.resolve is host-only → zero host→sandbox execs. expect(seen.get("workspace.resolve")?.payload?.roundTrips).toBe(0); @@ -2835,6 +2917,13 @@ describe("ACPX engine per-step startup timing (run.startup.step events)", () => expect(stageSync?.payload?.providerGetMs).toBe( (stageSync?.payload?.roundTrips as number) * 15, ); + // Concurrent bridge steps are duration-only so they do not double-count + // shared runner counters while their lifecycles overlap. + for (const step of ["bridge.paperclip", "bridge.process-session"]) { + expect(seen.get(step)?.payload?.roundTrips).toBeUndefined(); + expect(seen.get(step)?.payload?.providerExecMs).toBeUndefined(); + expect(seen.get(step)?.payload?.providerGetMs).toBeUndefined(); + } // The external ACP client crosses no host exec seam. expect(seen.get("acp.handshake")?.payload?.roundTrips).toBe(0); }); diff --git a/packages/adapter-utils/src/acpx-engine/execute.ts b/packages/adapter-utils/src/acpx-engine/execute.ts index 5c1ec5937e..67c4aae5a2 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.ts @@ -1292,6 +1292,12 @@ async function buildRuntime(input: { ? executionTarget.runner : undefined, ); + // The two bridge-start steps intentionally overlap, so their runner counters + // would double-count each other if we sampled them here. Keep the shared + // counter attribution on the sequential startup phases only; the concurrent + // bridge steps still emit duration telemetry, just not misleading per-step + // round-trip/provider deltas. + const concurrentBridgeStepMetrics: StartupStepMeasureOptions = {}; const shapedWorkspaceEnv = shapePaperclipWorkspaceEnvForExecution({ workspaceCwd: effectiveWorkspaceCwd, workspaceWorktreePath, @@ -1747,8 +1753,21 @@ async function buildRuntime(input: { let runtimeEnv: Record = {}; try { if (useRemoteProcessSession) { - // Step 5 — bridge.paperclip: start the sandbox ACP API callback bridge. - paperclipBridge = await measureStartupStep(input.ctx, nowMs, "bridge.paperclip", () => + // Steps 5 + 6 — bring up BOTH host-side sandbox bridges concurrently. Their + // remote subtrees are disjoint (`…/paperclip-bridge/…` vs + // `…/process-sessions/…`), so the env-INDEPENDENT setup of each overlaps, + // trending wall time from serial (~bridge.paperclip + ~bridge.process-session) + // toward ~max(the two). The ONE real dependency — the paperclip bridge's + // returned `env` must reach the process-session LAUNCH — is sequenced by + // `finalizeLaunchEnv`: the process-session bridge runs its env-independent + // dir/script setup first, then awaits that thunk right before its launch, so + // the launch always observes the merged paperclip env. + // + // Measurement caveat: both starts share ONE runner counter, so their + // overlapping `providerExecMs`/`roundTrips` deltas are approximate (the same + // caveat as `acp.handshake`). Both `run.startup.step` events still emit — + // `measureStartupStep` records them in a `finally`, even on a start failure. + const paperclipStart = measureStartupStep(input.ctx, nowMs, "bridge.paperclip", () => startAdapterExecutionTargetPaperclipBridge({ runId, target: { ...executionTarget, streamRunLogs: false }, @@ -1758,38 +1777,59 @@ async function buildRuntime(input: { hostApiToken: env.PAPERCLIP_API_KEY, onLog: input.ctx.onLog, }), - stepMetrics, + concurrentBridgeStepMetrics, ); - if (paperclipBridge) { - Object.assign(env, paperclipBridge.env); - await input.ctx.onLog("stdout", "[paperclip] Sandbox ACP API callback bridge enabled for this run.\n"); - } + // The single sequencing point (paperclip `env` → process-session launch). + // Memoized so the merge + log + `runtimeEnv` build run EXACTLY once whether + // the process-session bridge consumes it at launch or we finalize it below. + let launchEnvPromise: Promise> | null = null; + const finalizeLaunchEnv = (): Promise> => + (launchEnvPromise ??= (async () => { + const paperclip = await paperclipStart; + if (paperclip) { + Object.assign(env, paperclip.env); + await input.ctx.onLog("stdout", "[paperclip] Sandbox ACP API callback bridge enabled for this run.\n"); + } + return (runtimeEnv = resolveRuntimeEnv(env)); + })()); + const processSessionStart = measureStartupStep(input.ctx, nowMs, "bridge.process-session", () => + startAdapterExecutionTargetProcessSessionBridge({ + runId, + target: executionTarget, + runtimeRootDir: stagedRuntime?.runtimeRootDir ?? null, + adapterKey: input.engine.adapterType, + command: "sh", + args: ["-lc", `exec ${agentCommandShell}`], + cwd: sessionCwd, + // Deferred: the process-session bridge runs its env-independent setup, + // then calls this to get the launch env AFTER the paperclip env merge. + env: finalizeLaunchEnv, + timeoutSec, + onLog: input.ctx.onLog, + }), + concurrentBridgeStepMetrics, + ); + // Settle BOTH starts (mirrors `cleanupRemoteBridges`' `Promise.allSettled`): + // collect whichever handles started plus the first failure. Both handles + // stay individually declared so the catch below can stop whichever started. + const started = await settleRemoteBridgeStarts(paperclipStart, processSessionStart); + paperclipBridge = started.paperclipBridge; + processSessionBridge = started.processSessionBridge; + if (started.failure) throw started.failure; + // Guarantee the paperclip env merge ran even if the process-session bridge + // returned without consuming the launch env (memoized ⇒ a no-op if it did). + await finalizeLaunchEnv(); + } else { + // Local / runner-less lanes never start a bridge, but the returned prepared + // runtime and the log builder still read `runtimeEnv`. + runtimeEnv = resolveRuntimeEnv(env); } - runtimeEnv = Object.fromEntries( - Object.entries(ensurePathInEnv({ ...process.env, ...env })).filter( - (entry): entry is [string, string] => typeof entry[1] === "string", - ), - ); - // Step 6 — bridge.process-session: start the in-sandbox process session. - processSessionBridge = useRemoteProcessSession - ? await measureStartupStep(input.ctx, nowMs, "bridge.process-session", () => - startAdapterExecutionTargetProcessSessionBridge({ - runId, - target: executionTarget, - runtimeRootDir: stagedRuntime?.runtimeRootDir ?? null, - adapterKey: input.engine.adapterType, - command: "sh", - args: ["-lc", `exec ${agentCommandShell}`], - cwd: sessionCwd, - env: runtimeEnv, - timeoutSec, - onLog: input.ctx.onLog, - }), - stepMetrics, - ) - : null; } catch (err) { - await paperclipBridge?.stop().catch(() => {}); + // On a partial concurrent bring-up failure, ONE bridge may have started while + // the other threw; `Promise.allSettled` stops whichever started so no live + // bridge leaks (mirrors `cleanupRemoteBridges`). Both handles are individually + // declared above, so either may be non-null here. + await Promise.allSettled([paperclipBridge?.stop(), processSessionBridge?.stop()]); // The staged home / copy-back teardown must run even if a bridge fails to // start after the workspace + managed home were already staged into the // sandbox, so a refreshed credential is copied back on this error path too. @@ -1916,6 +1956,54 @@ async function applySessionConfigOptions(input: { } } +/** + * Build the process-session launch env: the host env overlaid with the run's + * `env` (so the merged paperclip bridge vars win) and a guaranteed `PATH`, + * narrowed to string values. Shared by the remote concurrent bring-up and the + * local / runner-less lane so both resolve the runtime env identically. + */ +function resolveRuntimeEnv(env: Record): Record { + return Object.fromEntries( + Object.entries(ensurePathInEnv({ ...process.env, ...env })).filter( + (entry): entry is [string, string] => typeof entry[1] === "string", + ), + ); +} + +/** + * Bring up the two host-side sandbox bridges concurrently and settle both. + * + * Mirrors `cleanupRemoteBridges`' `Promise.allSettled` idiom (settle, not + * `Promise.all`): running BOTH starts to completion is what lets the caller STOP + * a bridge that DID start when its sibling threw — so a partial failure never + * leaks a live bridge. Returns whichever handles started plus the first failure + * (paperclip before process-session) for the caller to rethrow through the + * shared abandon path. + */ +async function settleRemoteBridgeStarts( + paperclipStart: Promise, + processSessionStart: Promise, +): Promise<{ + paperclipBridge: AdapterExecutionTargetPaperclipBridgeHandle | null; + processSessionBridge: AdapterExecutionTargetProcessSessionBridgeHandle | null; + failure: unknown; +}> { + const [paperclip, processSession] = await Promise.allSettled([ + paperclipStart, + processSessionStart, + ]); + return { + paperclipBridge: paperclip.status === "fulfilled" ? paperclip.value : null, + processSessionBridge: processSession.status === "fulfilled" ? processSession.value : null, + failure: + paperclip.status === "rejected" + ? paperclip.reason + : processSession.status === "rejected" + ? processSession.reason + : null, + }; +} + async function cleanupRemoteBridges(prepared: AcpxPreparedRuntime): Promise { await Promise.allSettled([ prepared.processSessionBridge?.stop(), diff --git a/packages/adapter-utils/src/execution-target.ts b/packages/adapter-utils/src/execution-target.ts index b0a93bed05..3bf072f552 100644 --- a/packages/adapter-utils/src/execution-target.ts +++ b/packages/adapter-utils/src/execution-target.ts @@ -1303,7 +1303,12 @@ export async function startAdapterExecutionTargetProcessSessionBridge(input: { command: string; args: string[]; cwd: string; - env: Record; + // The launch env is consumed ONLY when building the base64 `commandPayload` + // below — never during the env-INDEPENDENT dir/script setup. Accepting a + // resolver (in addition to a plain object) lets a caller overlap that setup + // with other work — e.g. starting the paperclip callback bridge — and hand the + // merged env in right before the launch. + env: Record | (() => Promise>); timeoutSec?: number | null; onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise; }): Promise { @@ -1339,11 +1344,15 @@ export async function startAdapterExecutionTargetProcessSessionBridge(input: { await client.makeDir(eventsDir); await syncProcessSessionRemoteScript({ client, remoteScriptPath }); + // Resolve the launch env AFTER the env-independent setup above, so a caller + // can defer it until an upstream dependency (e.g. the paperclip bridge's env) + // is ready without blocking the dir/script setup. + const launchEnv = typeof input.env === "function" ? await input.env() : input.env; const commandPayload = Buffer.from(JSON.stringify({ command: input.command, args: input.args, cwd: input.cwd || target.remoteCwd, - env: sanitizeRemoteExecutionEnv(input.env), + env: sanitizeRemoteExecutionEnv(launchEnv), }), "utf8").toString("base64"); await onLog("stdout", `[paperclip] Starting ACP process session bridge in sandbox (${target.providerKey ?? "provider"}).\n`);