diff --git a/packages/adapter-utils/src/acpx-engine/startup-timing.test.ts b/packages/adapter-utils/src/acpx-engine/startup-timing.test.ts index 91255745d8..fa07faff41 100644 --- a/packages/adapter-utils/src/acpx-engine/startup-timing.test.ts +++ b/packages/adapter-utils/src/acpx-engine/startup-timing.test.ts @@ -7,6 +7,7 @@ import { getActiveStepContext, measureStartupStep, normalizeProviderFamily, + runWithoutActiveStep, SANDBOX_STARTUP_SPAN_ATTR_PREFIX, SANDBOX_STARTUP_SPAN_ATTRS, setSandboxRootSpanAttributes, @@ -495,6 +496,86 @@ describe("getActiveStepContext", () => { }, { criticalPath: false }); expect(seen!.criticalPath).toBe(false); }); + + it("keeps the ended step store in a timer scheduled inside the step body", async () => { + // Node snapshots the active store on each async resource at creation time. + // A timer scheduled inside a measured step body keeps that step store, even + // after the step span ends. A later run-time exec then reads the ended step + // and parents its `sandbox.exec` span to a dead startup step. This test locks + // that leak, so the fix and its guard test stay honest. + let storeInTimer: ReturnType = null; + let fireTimer!: () => void; + const timerRan = new Promise((resolve) => { + fireTimer = resolve; + }); + + await measureStartupStep( + { onEvent: vi.fn(async () => {}) }, + () => 0, + "bridge.process-session", + async () => { + setTimeout(() => { + storeInTimer = getActiveStepContext(); + fireTimer(); + }, 0); + return "started"; + }, + { criticalPath: false }, + ); + + // The step span already ended, so the main line reads no active step. + expect(getActiveStepContext()).toBeNull(); + + await timerRan; + + // Yet the timer callback still reads the ended step context. This is the + // store leak the bridge boundary fix removes. + expect(storeInTimer).not.toBeNull(); + expect(storeInTimer!.criticalPath).toBe(false); + }); + + it("clears the active step for a continuation wrapped in runWithoutActiveStep", async () => { + // The bridge boundary wraps its long-lived poll timer in `runWithoutActiveStep`. + // A timer scheduled inside that empty store scope reads no active step, so a + // later run-time exec opens an unparented span instead of one under the ended + // startup step. + let storeInTimer: ReturnType = null; + let sawTimer = false; + let fireTimer!: () => void; + const timerRan = new Promise((resolve) => { + fireTimer = resolve; + }); + + await measureStartupStep( + { onEvent: vi.fn(async () => {}) }, + () => 0, + "bridge.process-session", + async () => { + runWithoutActiveStep(() => { + setTimeout(() => { + storeInTimer = getActiveStepContext(); + sawTimer = true; + fireTimer(); + }, 0); + }); + return "started"; + }, + { criticalPath: false }, + ); + + await timerRan; + + expect(sawTimer).toBe(true); + expect(storeInTimer).toBeNull(); + }); + + it("returns the work result and restores the previous active step", async () => { + // Outside any measured step the previous store is empty, so the helper both + // returns the work value and leaves the store empty afterward. + const value = runWithoutActiveStep(() => "value"); + expect(value).toBe("value"); + expect(getActiveStepContext()).toBeNull(); + }); }); describe("clampSpanLabel", () => { diff --git a/packages/adapter-utils/src/acpx-engine/startup-timing.ts b/packages/adapter-utils/src/acpx-engine/startup-timing.ts index 1de2ceeb88..d13ba73659 100644 --- a/packages/adapter-utils/src/acpx-engine/startup-timing.ts +++ b/packages/adapter-utils/src/acpx-engine/startup-timing.ts @@ -332,7 +332,7 @@ export interface ActiveStepContext { * a module-level singleton, so the value propagates across `await` boundaries * and across package boundaries that share this module. */ -const activeStepContextStorage = new AsyncLocalStorage(); +const activeStepContextStorage = new AsyncLocalStorage(); /** * Return the active step context, or `null` when no measured step is running. @@ -344,6 +344,26 @@ export function getActiveStepContext(): ActiveStepContext | null { return activeStepContextStorage.getStore() ?? null; } +/** + * Run `work` with no active step context, then restore the previous store. A + * bridge boundary uses this to start its long-lived poll timer and socket + * handlers outside the measured step store. + * + * Node snapshots the active store on each async resource at creation time. So a + * timer or a handler scheduled inside a measured step body keeps that step store + * after the step span ends. A later run-time exec then reads the ended step and + * parents its `sandbox.exec` span to a dead startup step, and it copies the + * step's `criticalPath` flag. This helper resets the store for the wrapped work, + * so each continuation reads an empty store. Each run-time exec then opens an + * unparented span with no stale `criticalPath` flag. + * + * The helper forwards only the opaque store, so this package stays free of + * `@opentelemetry/api`. It needs no Node version gate. + */ +export function runWithoutActiveStep(work: () => T): T { + return activeStepContextStorage.run(undefined, work); +} + /** * Set a numeric span attribute only when the value is a finite number. A reader * that returns `undefined` (the counter is unavailable) yields no attribute, diff --git a/packages/adapter-utils/src/execution-target.ts b/packages/adapter-utils/src/execution-target.ts index d1d878d7c2..2ee9c0f049 100644 --- a/packages/adapter-utils/src/execution-target.ts +++ b/packages/adapter-utils/src/execution-target.ts @@ -46,6 +46,7 @@ import { } from "./server-utils.js"; import { sanitizeRemoteExecutionEnv } from "./remote-execution-env.js"; import { preferredShellForSandbox, shellCommandArgs } from "./sandbox-shell.js"; +import { runWithoutActiveStep } from "./acpx-engine/startup-timing.js"; import type { RuntimeProgressSink, RuntimeStatusSink } from "./runtime-progress.js"; import type { LocalProcessSandboxOptions } from "./local-process-sandbox.js"; @@ -1488,7 +1489,10 @@ export async function startAdapterExecutionTargetProcessSessionBridge(input: { }; const liveSockets = new Set(); - const server = net.createServer((nextSocket) => { + // Register the per-connection socket handlers outside the measured bridge step + // store. A stdin write from a socket handler is a run-time exec, not startup + // work, so its `sandbox.exec` span must not parent to the ended bridge step. + const server = net.createServer((nextSocket) => runWithoutActiveStep(() => { liveSockets.add(nextSocket); nextSocket.setEncoding("utf8"); nextSocket.on("error", () => undefined); @@ -1547,7 +1551,7 @@ export async function startAdapterExecutionTargetProcessSessionBridge(input: { }); } }); - }); + })); const poll = async () => { if (stopping) return; @@ -1572,16 +1576,24 @@ export async function startAdapterExecutionTargetProcessSessionBridge(input: { return; } finally { if (!stopping) { - pollTimer = setTimeout(() => void poll(), 100); - pollTimer.unref?.(); + schedulePoll(); } } }; + // Schedule the long-lived poll timer outside the measured bridge step store. + // The poll loop reads remote event files with run-time execs, not startup + // work, so a poll `sandbox.exec` span must not parent to the ended bridge step. + // `runWithoutActiveStep` also empties the store for the re-arm timer that the + // poll body schedules, so every later tick stays unparented too. + const schedulePoll = () => { + pollTimer = setTimeout(() => runWithoutActiveStep(() => void poll()), 100); + pollTimer.unref?.(); + }; + const port = await waitForLocalServerListen(server); const agentCommand = await writeProcessSessionProxyScript(proxyDir, port, token); - pollTimer = setTimeout(() => void poll(), 100); - pollTimer.unref?.(); + schedulePoll(); return { agentCommand, @@ -1787,6 +1799,11 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: { // this flag is enabled. Only intended for active debugging in trusted // environments. const bridgeDebugEnabled = isBridgeDebugEnabled(process.env); + // `startSandboxCallbackBridgeWorker` keeps its awaited queue-directory + // setup on the active `bridge.paperclip` step, and resets the store only + // for its long-lived poll loop (see `runWithoutActiveStep` inside that + // function). So the startup `mkdir` execs stay parented and every later + // loop `sandbox.exec` span stays unparented with no stale `criticalPath`. worker = await startSandboxCallbackBridgeWorker({ client, queueDir, diff --git a/packages/adapter-utils/src/sandbox-callback-bridge.test.ts b/packages/adapter-utils/src/sandbox-callback-bridge.test.ts index 73fc3b472a..f36f63bbef 100644 --- a/packages/adapter-utils/src/sandbox-callback-bridge.test.ts +++ b/packages/adapter-utils/src/sandbox-callback-bridge.test.ts @@ -5,6 +5,7 @@ import path from "node:path"; import { promisify } from "node:util"; import { afterEach, describe, expect, it, vi } from "vitest"; +import { getActiveStepContext, measureStartupStep } from "./acpx-engine/startup-timing.js"; import { prepareCommandManagedRuntime } from "./command-managed-runtime.js"; import { authorizeSandboxCallbackBridgeRequestWithRoutes, @@ -471,6 +472,73 @@ describe("sandbox callback bridge", () => { } }); + it("keeps the queue-directory setup on the startup step but resets the poll loop store", async () => { + // The worker starts inside the measured `bridge.paperclip` step. Its awaited + // queue-directory setup is startup work, so a `makeDir` `sandbox.exec` span + // must keep the active step and its `criticalPath` flag. The long-lived poll + // loop runs run-time execs for the whole run, so a loop `sandbox.exec` span + // must open unparented with no stale flag. This test reads the active step in + // both places and proves the boundary sits at the loop, not the whole worker. + let setupStep: ReturnType | "unset" = "unset"; + let loopStep: ReturnType | "unset" = "unset"; + let resolveFirstPoll: () => void = () => {}; + const firstPoll = new Promise((resolve) => { + resolveFirstPoll = resolve; + }); + + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-bridge-step-store-")); + cleanupDirs.push(rootDir); + const queueDir = path.posix.join(rootDir, "queue"); + + const worker = await measureStartupStep( + {}, + () => 0, + "bridge.paperclip", + () => + startSandboxCallbackBridgeWorker({ + client: { + makeDir: async () => { + setupStep = getActiveStepContext(); + }, + makeDirs: async () => { + setupStep = getActiveStepContext(); + }, + listJsonFiles: async () => { + loopStep = getActiveStepContext(); + resolveFirstPoll(); + return []; + }, + readTextFile: async () => { + throw new Error("unexpected readTextFile"); + }, + writeTextFile: async () => { + throw new Error("unexpected writeTextFile"); + }, + rename: async () => { + throw new Error("unexpected rename"); + }, + remove: async () => {}, + }, + queueDir, + authorizeRequest: async () => null, + handleRequest: async () => ({ status: 200, body: "ok" }), + }), + { criticalPath: false }, + ); + + await firstPoll; + await worker.stop(); + + // The setup ran on the active step, so its exec span parents to the step. + expect(setupStep).not.toBe("unset"); + expect(setupStep).not.toBeNull(); + expect((setupStep as { criticalPath?: boolean }).criticalPath).toBe(false); + + // The loop ran outside that store, so its exec span opens unparented with no + // stale `criticalPath` flag. + expect(loopStep).toBeNull(); + }); + it("serializes remote response writes so stop does not recreate a late orphaned response", async () => { const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-bridge-response-lock-")); cleanupDirs.push(rootDir); diff --git a/packages/adapter-utils/src/sandbox-callback-bridge.ts b/packages/adapter-utils/src/sandbox-callback-bridge.ts index d990adbef3..b21d8a1874 100644 --- a/packages/adapter-utils/src/sandbox-callback-bridge.ts +++ b/packages/adapter-utils/src/sandbox-callback-bridge.ts @@ -3,6 +3,7 @@ import { promises as fs } from "node:fs"; import os from "node:os"; import path from "node:path"; +import { runWithoutActiveStep } from "./acpx-engine/startup-timing.js"; import type { CommandManagedRuntimeRunner } from "./command-managed-runtime.js"; import { preferredShellForSandbox, shellCommandArgs } from "./sandbox-shell.js"; import type { RunProcessResult } from "./server-utils.js"; @@ -745,7 +746,13 @@ export async function startSandboxCallbackBridgeWorker(input: { } }; - const loop = (async () => { + // Start the long-lived poll loop outside the measured startup-step store. + // The `makeDir` calls above are startup work and must keep the active + // `bridge.paperclip` step. The loop runs run-time execs for the whole run, + // so each loop `sandbox.exec` span must not parent to the ended step or copy + // its `criticalPath` flag. `runWithoutActiveStep` empties the store for the + // loop only; Node keeps the empty store on every later poll continuation. + const loop = runWithoutActiveStep(() => (async () => { try { while (true) { const fileNames = await input.client.listJsonFiles(directories.requestsDir); @@ -785,7 +792,7 @@ export async function startSandboxCallbackBridgeWorker(input: { settleResolve(); } } - })(); + })()); void loop; diff --git a/server/src/__tests__/environment-execution-target.test.ts b/server/src/__tests__/environment-execution-target.test.ts index e59e6022e7..c49557dc71 100644 --- a/server/src/__tests__/environment-execution-target.test.ts +++ b/server/src/__tests__/environment-execution-target.test.ts @@ -10,6 +10,7 @@ vi.mock("../services/environment-config.js", () => ({ import { measureStartupStep, + runWithoutActiveStep, SANDBOX_STARTUP_SPAN_ATTRS, } from "@paperclipai/adapter-utils/acpx-engine/startup-timing"; import { @@ -847,4 +848,116 @@ describe("resolveEnvironmentExecutionTarget", () => { // The seam reached the log callback exactly once (the stdout delivery). expect(onLog).toHaveBeenCalledTimes(1); }); + + // Fire one run-time exec from a bridge continuation that runs after the step + // span ended. Each bridge step (`bridge.paperclip`, `bridge.process-session`) + // starts long-lived work with `criticalPath: false`. The bridge boundary wraps + // that long-lived work in `runWithoutActiveStep`, exactly as modeled here, so + // the continuation reads an empty active step. Return the recorded exec span. + async function runContinuationExec(step: string, options: { wrap: boolean }) { + const { tracer, contextWithSpan, spans } = createRecordingTrace(); + const runner = await runnerFor({ + provider: "daytona", + execResult: { exitCode: 0, signal: null, timedOut: false, stdout: "", stderr: "" }, + tracer, + }); + + let resolveExec!: () => void; + const execDone = new Promise((resolve) => { + resolveExec = resolve; + }); + + // Schedule the exec from a timer inside the step body, so it fires after the + // step span ends. The `wrap` flag models the fix: when true, the boundary + // wraps the long-lived work in `runWithoutActiveStep`; when false, it models + // the pre-fix leak. + const scheduleContinuation = () => { + setTimeout(() => { + void runner.execute({ command: "echo" }).then(() => resolveExec()); + }, 0); + }; + + await measureStartupStep( + {}, + () => 0, + step, + async () => { + if (options.wrap) { + runWithoutActiveStep(scheduleContinuation); + } else { + scheduleContinuation(); + } + return "started"; + }, + { tracer, contextWithSpan, criticalPath: false }, + ); + + await execDone; + return spans.find((span) => span.name === "sandbox.exec"); + } + + it("opens an unparented exec span for a process-session bridge continuation", async () => { + const execSpan = await runContinuationExec("bridge.process-session", { wrap: true }); + expect(execSpan).toBeTruthy(); + // The step span ended and the boundary emptied the store, so the continuation + // exec opens a root span, not one under the dead bridge step. + expect(execSpan!.parent).toBeNull(); + }); + + it("opens an unparented exec span for a paperclip bridge continuation", async () => { + const execSpan = await runContinuationExec("bridge.paperclip", { wrap: true }); + expect(execSpan).toBeTruthy(); + expect(execSpan!.parent).toBeNull(); + }); + + it("does not copy the stale criticalPath = false flag onto a continuation exec", async () => { + const execSpan = await runContinuationExec("bridge.process-session", { wrap: true }); + expect(execSpan).toBeTruthy(); + // The bridge step set `criticalPath: false`. The continuation reads an empty + // store, so the exec span records the default `true`, never the stale `false`. + expect(execSpan!.attributes[A.execCriticalPath]).toBe(true); + expect(execSpan!.attributes[A.execCriticalPath]).not.toBe(false); + }); + + it("leaks the ended step onto a continuation exec without the boundary wrap", async () => { + // The mechanism guard: an unwrapped continuation keeps the ended bridge step + // store, so the exec span parents to the dead step and copies its + // `criticalPath: false`. The boundary wrap in the two tests above removes both + // defects, so this suite fails if a future edit drops the wrap. + const { tracer, contextWithSpan, spans } = createRecordingTrace(); + const runner = await runnerFor({ + provider: "daytona", + execResult: { exitCode: 0, signal: null, timedOut: false, stdout: "", stderr: "" }, + tracer, + }); + + let resolveExec!: () => void; + const execDone = new Promise((resolve) => { + resolveExec = resolve; + }); + + await measureStartupStep( + {}, + () => 0, + "bridge.process-session", + async () => { + setTimeout(() => { + void runner.execute({ command: "echo" }).then(() => resolveExec()); + }, 0); + return "started"; + }, + { tracer, contextWithSpan, criticalPath: false }, + ); + + await execDone; + + const stepSpan = spans.find((span) => span.name === "bridge.process-session"); + const execSpan = spans.find((span) => span.name === "sandbox.exec"); + expect(stepSpan).toBeTruthy(); + expect(execSpan).toBeTruthy(); + // The unwrapped continuation parents the exec span to the ended step and + // copies the stale flag. + expect(execSpan!.parent).toBe(stepSpan); + expect(execSpan!.attributes[A.execCriticalPath]).toBe(false); + }); });