refactor(eve): reuse the tool sandbox accessor in workflow steps

Signed-off-by: Rui Conti <ruiconti@gmail.com>
2026-09-20 05:35:39 +08:00 · 2026-09-19 14:18:27 -04:00
parent 1769333c4b
commit 3b01864f2c
8 changed files with 242 additions and 132 deletions
@@ -105,8 +105,9 @@ the build with the missing import in the error.

 Call `ctx.getSandbox()` in an authored step to open or reconnect the session sandbox. Both
 blocking and background tools support this accessor without configuration. Workflows that
-never call it do not open a sandbox. Each step reconnects to the same sandbox; files written
-by an earlier step remain available to later steps.
+never call it do not open a sandbox. Regular tools and workflow steps use the same session
+sandbox. Each step reconstructs access from the session's saved state, so files remain available
+across tools and later steps.

 ```ts title="agent/tools/read_report.ts"
 import { defineWorkflowTool, type WorkflowStepToolContext } from "eve/tools";
@@ -1,5 +1,5 @@
-import { describe, expect, it, vi } from "vitest";
-import { openWorkflowSandboxStep } from "#execution/sandbox/workflow-session-step.js";
+import { beforeEach, describe, expect, it, vi } from "vitest";
+import { createWorkflowSandboxAccess } from "#execution/sandbox/workflow-session-step.js";
 import { mockSandbox } from "#internal/testing/mocks/mock-sandbox.js";
 import type { WorkflowSandboxReferenceData } from "#execution/sandbox/workflow-reference.js";

@@ -8,51 +8,75 @@ vi.mock("#execution/sandbox/ensure.js", () => ({ ensureSandboxAccess: mocks.ensu
 vi.mock("#runtime/sessions/compiled-agent-cache.js", () => ({
  getCompiledRuntimeAgentBundle: mocks.bundle,
 }));
-
 vi.mock("#execution/sandbox/workflow-request.js", () => ({
  requestWorkflowSandbox: mocks.request,
 }));

+const reference: WorkflowSandboxReferenceData = {
+  compiledArtifactsSource: { kind: "bundled" },
+  nodeId: "root",
+  sessionId: "parent-session",
+  state: { initialized: true, session: null },
+};
+const run = {
+  owner: { inbox: "owner-inbox" },
+  from: {
+    callId: "call-1",
+    execution: "blocking" as const,
+    input: {},
+    runId: "run-1",
+    sequence: 0,
+    stepIndex: 0,
+    toolName: "probe",
+    turnId: "turn-1",
+  },
+};
+const registry = { sandbox: null };
+
+function createAccess() {
+  return createWorkflowSandboxAccess({ run, abortSignal: new AbortController().signal });
+}
+
 describe("workflow sandbox access", () => {
-  it("borrows the recorded sandbox, binds cancellation, and forbids lifecycle mutations", async () => {
-    const sandbox = mockSandbox();
-    const run = vi.spyOn(sandbox.session, "run");
-    const stop = vi.fn();
-    const remove = vi.fn();
-    mocks.ensure.mockResolvedValue({ ...sandbox.access, stop, delete: remove });
-    const registry = { sandbox: null };
+  beforeEach(() => {
+    vi.resetAllMocks();
    mocks.bundle.mockResolvedValue({ graph: { root: { sandboxRegistry: registry } } });
-    const reference: WorkflowSandboxReferenceData = {
-      compiledArtifactsSource: { kind: "bundled" },
-      nodeId: "root",
-      sessionId: "parent-session",
-      state: { initialized: true, session: null },
-    };
    mocks.request.mockResolvedValue(reference);
-    const context = {
-      owner: { inbox: "owner-inbox" },
-      from: {
-        callId: "call-1",
-        execution: "blocking" as const,
-        input: {},
-        runId: "run-1",
-        sequence: 0,
-        stepIndex: 0,
-        toolName: "probe",
-        turnId: "turn-1",
-      },
-    };
-    const controller = new AbortController();
-    const handle = await openWorkflowSandboxStep({ run: context, abortSignal: controller.signal });
-    expect(mocks.ensure).toHaveBeenCalledWith({ ...reference, ownsSandbox: false, registry });
-    await handle.run({ command: "echo ready" });
-    const signal = run.mock.calls[0]?.[0].abortSignal;
-    expect(signal?.aborted).toBe(false);
-    controller.abort();
-    expect(signal?.aborted).toBe(true);
-    expect(() => handle.stop()).toThrow("session owns its lifecycle");
-    expect(() => handle.delete()).toThrow("not available");
-    expect(stop).not.toHaveBeenCalled();
-    expect(remove).not.toHaveBeenCalled();
+  });
+
+  it("opens lazily and shares access between concurrent callers in one step", async () => {
+    const sandbox = mockSandbox();
+    mocks.ensure.mockResolvedValue(sandbox.access);
+    const access = createAccess();
+    expect(mocks.request).not.toHaveBeenCalled();
+    await access.captureState();
+    expect(mocks.request).not.toHaveBeenCalled();
+    const [first, second] = await Promise.all([access.get(), access.get()]);
+    expect(first).toBe(sandbox.session);
+    expect(second).toBe(first);
+    expect(mocks.request).toHaveBeenCalledOnce();
+    expect(mocks.ensure).toHaveBeenCalledExactlyOnceWith({
+      ...reference,
+      ownsSandbox: false,
+      registry,
+    });
+  });
+
+  it("reconstructs access from the owner's state for a new step context", async () => {
+    const sandbox = mockSandbox();
+    mocks.ensure.mockImplementation(async () => ({ ...sandbox.access }));
+    const first = createAccess();
+    const second = createAccess();
+    expect(first).not.toBe(second);
+    expect(await first.get()).toBe(await second.get());
+    expect(mocks.ensure).toHaveBeenCalledTimes(2);
+    for (const [input] of mocks.ensure.mock.calls) expect(input.state).toBe(reference.state);
+  });
+
+  it("leaves sandbox lifecycle mutations with the owning session", async () => {
+    const access = createAccess();
+    await expect(access.stop()).rejects.toThrow("session owns its lifecycle");
+    await expect(access.delete!()).rejects.toThrow("not available");
+    expect(mocks.request).not.toHaveBeenCalled();
  });
 });
@@ -1,59 +1,51 @@
-import { bindSandboxAbortSignal } from "#execution/sandbox/abort-bound-session.js";
 import { ensureSandboxAccess } from "#execution/sandbox/ensure.js";
 import type { WorkflowToolRunContext } from "#execution/tools/workflow/ask.js";
 import { requestWorkflowSandbox } from "#execution/sandbox/workflow-request.js";
 import { getCompiledRuntimeAgentBundle } from "#runtime/sessions/compiled-agent-cache.js";
-import type { RuntimeSandboxSession, SandboxSession } from "#shared/sandbox-session.js";
+import type { SandboxAccess } from "#sandbox/state.js";

-export async function openWorkflowSandboxStep(input: {
+export function createWorkflowSandboxAccess(input: {
  readonly abortSignal: AbortSignal;
-  readonly run: WorkflowToolRunContext;
-}): Promise<RuntimeSandboxSession> {
-  const reference = await requestWorkflowSandbox(input);
-  const bundle = await getCompiledRuntimeAgentBundle({
-    compiledArtifactsSource: reference.compiledArtifactsSource,
-    nodeId: reference.nodeId,
-  });
-  const access = await ensureSandboxAccess({
-    ...reference,
-    ownsSandbox: false,
-    registry: bundle.graph.root.sandboxRegistry,
-  });
-  const sandbox = await access.get();
-  if (sandbox === null) {
-    throw new Error("The sandbox is not available in the current authored runtime context.");
+  readonly run?: WorkflowToolRunContext;
+}): SandboxAccess {
+  let access: Promise<SandboxAccess> | undefined;
+  async function open(): Promise<SandboxAccess> {
+    if (input.run === undefined) {
+      throw Object.assign(
+        new Error(
+          'ctx.getSandbox() is unavailable inside a "use step" function. Pass the workflow context directly to this step.',
+        ),
+        { fatal: true },
+      );
+    }
+    const reference = await requestWorkflowSandbox({ ...input, run: input.run });
+    const bundle = await getCompiledRuntimeAgentBundle({
+      compiledArtifactsSource: reference.compiledArtifactsSource,
+      nodeId: reference.nodeId,
+    });
+    return ensureSandboxAccess({
+      ...reference,
+      ownsSandbox: false,
+      registry: bundle.graph.root.sandboxRegistry,
+    });
  }
-  return bindSandboxAbortSignal(
-    withWorkflowSandboxLifecycle({
-      sandbox,
-    }),
-    input.abortSignal,
-  );
-}

-function withWorkflowSandboxLifecycle(input: {
-  readonly sandbox: SandboxSession;
-}): RuntimeSandboxSession {
  return {
-    delete() {
+    async get() {
+      return (await (access ??= open())).get();
+    },
+    async captureState() {
+      return access === undefined
+        ? { initialized: false, session: null }
+        : (await access).captureState();
+    },
+    async delete() {
      throw new Error("sandbox.delete() is not available inside a defineWorkflowTool() step.");
    },
-    id: input.sandbox.id,
-    readBinaryFile: (options) => input.sandbox.readBinaryFile(options),
-    readFile: (options) => input.sandbox.readFile(options),
-    readTextFile: (options) => input.sandbox.readTextFile(options),
-    removePath: (options) => input.sandbox.removePath(options),
-    resolvePath: (path) => input.sandbox.resolvePath(path),
-    run: (options) => input.sandbox.run(options),
-    setNetworkPolicy: (policy) => input.sandbox.setNetworkPolicy(policy),
-    spawn: (options) => input.sandbox.spawn(options),
-    stop() {
+    async stop() {
      throw new Error(
        "sandbox.stop() is unavailable inside a workflow step; the session owns its lifecycle.",
      );
    },
-    writeBinaryFile: (options) => input.sandbox.writeBinaryFile(options),
-    writeFile: (options) => input.sandbox.writeFile(options),
-    writeTextFile: (options) => input.sandbox.writeTextFile(options),
  };
 }
@@ -2,7 +2,7 @@ import { mockSandbox } from "#internal/testing/mocks/mock-sandbox.js";
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import { withWorkflowStepAuthorization } from "#execution/tools/workflow/step-execution.js";
 import { ContextContainer, contextStorage } from "#context/container.js";
-import { AuthKey } from "#context/keys.js";
+import { AuthKey, SandboxKey } from "#context/keys.js";
 import {
  ConnectionAuthorizationRequiredError,
  ConnectionAuthorizationFailedError,
@@ -40,9 +40,9 @@ vi.mock("#internal/workflow/runtime.js", () => ({
  }),
 }));

-const openSandbox = vi.hoisted(() => vi.fn());
-vi.mock("#execution/sandbox/workflow-session-step.js", () => ({
-  openWorkflowSandboxStep: openSandbox,
+const requestSandbox = vi.hoisted(() => vi.fn());
+vi.mock("#execution/sandbox/workflow-request.js", () => ({
+  requestWorkflowSandbox: requestSandbox,
 }));

 function context(user = "user-1"): WorkflowStepContext {
@@ -86,41 +86,42 @@ describe("workflow step authorization", () => {
    durable.entries.clear();
  });
  afterEach(() => vi.unstubAllEnvs());
-  it("rejects sandbox access without a workflow owner before opening a backend", async () => {
-    openSandbox.mockClear();
-    await expect(runStep((ctx) => ctx.getSandbox())).rejects.toMatchObject({ fatal: true });
-    expect(openSandbox).not.toHaveBeenCalled();
+  it("does not request a sandbox when the step never accesses it", async () => {
+    requestSandbox.mockClear();
+    await expect(runStep(() => "done")).resolves.toMatchObject({ output: "done" });
+    expect(requestSandbox).not.toHaveBeenCalled();
  });

-  it("reuses one sandbox handle within a step and forwards its owner identity", async () => {
+  it("rejects sandbox access without a workflow owner before opening a backend", async () => {
+    requestSandbox.mockClear();
+    await expect(runStep((ctx) => ctx.getSandbox())).rejects.toMatchObject({ fatal: true });
+    expect(requestSandbox).not.toHaveBeenCalled();
+  });
+
+  it("uses the context sandbox accessor and binds the tool's cancellation signal", async () => {
    const sandbox = mockSandbox();
-    openSandbox.mockResolvedValue(sandbox.session);
-    const input: WorkflowStepContext = {
-      ...context(),
-      run: {
-        owner: { inbox: "owner-inbox" },
-        from: {
-          callId: "call-1",
-          execution: "blocking" as const,
-          input: {},
-          runId: "run-1",
-          sequence: 0,
-          stepIndex: 0,
-          toolName: "probe",
-          turnId: "turn-1",
-        },
-      },
-    };
+    const run = vi.spyOn(sandbox.session, "run");
+    const controller = new AbortController();
+    const input = { ...context(), abortSignal: controller.signal };
    const result = await runStep(async (ctx) => {
+      contextStorage.getStore()!.setVirtualContext(SandboxKey, sandbox.access);
      const first = await ctx.getSandbox();
      const second = await ctx.getSandbox();
-      expect(second).toBe(first);
+      expect(second.id).toBe(first.id);
+      await first.run({ command: "echo ready" });
      return first.id;
    }, input);
    expect(result).toMatchObject({ kind: "result", output: sandbox.session.id });
-    expect(openSandbox).toHaveBeenCalledExactlyOnceWith({
-      abortSignal: input.abortSignal,
-      run: input.run,
+    const signal = run.mock.calls[0]?.[0].abortSignal;
+    expect(signal?.aborted).toBe(false);
+    controller.abort();
+    expect(signal?.aborted).toBe(true);
+  });
+
+  it("keeps getSkill unavailable in workflow steps", async () => {
+    await expect(runStep((ctx) => ctx.getSkill("example"))).rejects.toMatchObject({
+      fatal: true,
+      message: expect.stringContaining("ctx.getSkill()"),
    });
  });

@@ -1,7 +1,7 @@
-import { openWorkflowSandboxStep } from "#execution/sandbox/workflow-session-step.js";
+import { createWorkflowSandboxAccess } from "#execution/sandbox/workflow-session-step.js";
 import { getStepMetadata } from "#compiled/@workflow/core/index.js";
 import { ContextContainer, contextStorage } from "#context/container.js";
-import { AuthKey, InitiatorAuthKey, SessionIdKey, SessionKey } from "#context/keys.js";
+import { AuthKey, InitiatorAuthKey, SandboxKey, SessionIdKey, SessionKey } from "#context/keys.js";
 import { isConnectionAuthorizationFailedError } from "#connections/errors.js";
 import {
  isAuthorizationSignal,
@@ -34,13 +34,13 @@ export function withWorkflowStepAuthorization(execute: (...args: never[]) => unk
    context.set(CallbackBaseUrlKey, resolveWorkflowCallbackBaseUrl(input.baseUrl));
    context.setVirtualContext(AuthorizationHookKey, input.token);
    context.setVirtualContext(PendingAuthorizationResultKey, input.authorizationResults);
+    context.setVirtualContext(SandboxKey, createWorkflowSandboxAccess(input));

    return contextStorage.run(context, async (): Promise<WorkflowStepResult> => {
      const auth = createAuthorizationContext({
        scope: input.toolName,
        completeAuthorization: completeWorkflowStepAuthorization,
      });
-      let sandbox: ReturnType<typeof openWorkflowSandboxStep> | undefined;
      const ctx = {
        ...buildBaseToolContext({
          toolName: input.toolName,
@@ -48,18 +48,8 @@ export function withWorkflowStepAuthorization(execute: (...args: never[]) => unk
        }),
        agent: () => unavailableInStep("ctx.agent()", "Call ctx.agent() in the workflow body."),
        ask: () => unavailableInStep("ctx.ask()", "Call ctx.ask() in the workflow body."),
-        getSandbox: () => {
-          if (input.run === undefined) {
-            return unavailableInStep(
-              "ctx.getSandbox()",
-              "Pass the workflow context directly to this step.",
-            );
-          }
-          return (sandbox ??= openWorkflowSandboxStep({
-            abortSignal: input.abortSignal,
-            run: input.run,
-          }));
-        },
+        getSkill: () =>
+          unavailableInStep("ctx.getSkill()", "Read skill files through ctx.getSandbox()."),
        getToken: auth.getToken,
        requireAuth: auth.requireAuth,
      };
@@ -18,6 +18,7 @@ import {
  holdUntilAbortedWorkflow,
  reportingDeployWorkflow,
  sandboxAcrossStepsWorkflow,
+  sharedSandboxWorkflow,
  concurrentSandboxWorkflow,
  recoverSandboxFailureWorkflow,
  sandboxFromWorkflowBodyWorkflow,
@@ -36,6 +37,7 @@ import {
  defineWorkflowTool,
  type BlockingWorkflowToolDefinition,
 } from "#tools/workflow-definition.js";
+import { defineTool } from "#tools/definition.js";
 import { serializeInputSchema, toInputSchema } from "#tools/schema.js";

 const DEPLOY_INPUT_SCHEMA = toInputSchema({
@@ -523,6 +525,90 @@ describe("workflow tools", () => {
    expect(output).toContain("Attempt 1.");
  });

+  it.each([false, true])(
+    "shares sandbox state between regular tools and workflow steps (background=%s)",
+    async (background) => {
+      const runtime = await createTestRuntime({
+        modules: [
+          {
+            logicalPath: "tools/regular_probe.ts",
+            loadNamespace: async () => ({
+              default: defineTool({
+                description: "Seed or read the shared sandbox.",
+                inputSchema: serializeInputSchema(DEPLOY_INPUT_SCHEMA) ?? {},
+                async execute(input, ctx) {
+                  const sandbox = await ctx.getSandbox();
+                  if ((input as { service: string }).service === "seed") {
+                    await sandbox.writeTextFile({ path: "shared.txt", content: "regular" });
+                  }
+                  return {
+                    id: sandbox.id,
+                    content: await sandbox.readTextFile({ path: "shared.txt" }),
+                  };
+                },
+              }),
+            }),
+          },
+          {
+            logicalPath: "tools/sandbox_probe.ts",
+            loadNamespace: async () => ({
+              default: defineWorkflowTool({
+                description: "Update the regular tool's sandbox from separate steps.",
+                execution: background ? "background" : undefined,
+                inputSchema: serializeInputSchema(DEPLOY_INPUT_SCHEMA) ?? {},
+                execute: sharedSandboxWorkflow,
+              }),
+            }),
+          },
+        ],
+      });
+      await runtime.run(async () => {
+        const bundle = await getCompiledRuntimeAgentBundle({
+          compiledArtifactsSource: createBundledRuntimeCompiledArtifactsSource(),
+        });
+        const initialized = vi.fn(async () => {});
+        Object.assign(bundle.graph.root.sandboxRegistry.sandbox!.definition, {
+          onSession: initialized,
+        });
+        const run = await start(workflowEntry, [
+          {
+            kind: "initial",
+            ownerDeploymentId: "dpl_inline",
+            input: { message: 'Run regular_probe with service "seed"' },
+            serializedContext: buildSerializedContext({
+              continuationToken: "http:shared-sandbox",
+              mode: "conversation",
+            }),
+          },
+        ]);
+        const stream = captureTurnEvents(run);
+        const send = async (message: string) =>
+          resumeSessionInbox(sessionCommandHookToken(run.runId), {
+            kind: "send",
+            payload: { message },
+          });
+        try {
+          const seeded = JSON.stringify(await stream.nextTurn());
+          expect(seeded).toContain("regular");
+          await send('Run sandbox_probe with service "api"');
+          let workflow = "";
+          for (let turn = 0; turn < 4 && !workflow.includes("regular|workflow|workflow"); turn++) {
+            workflow += JSON.stringify(await stream.nextTurn());
+          }
+          expect(workflow).toContain("regular|workflow|workflow");
+          await send('Run regular_probe with service "read"');
+          const read = JSON.stringify(await stream.nextTurn());
+          expect(read).toContain("regular|workflow|workflow");
+          expect(initialized).toHaveBeenCalledTimes(1);
+        } finally {
+          stream.dispose();
+          await run.cancel();
+        }
+      });
+    },
+    60_000,
+  );
+
  it("reattaches the session sandbox across workflow steps", async () => {
    const runtime = await createWorkflowToolRuntime({
      agentName: "workflow-tool-sandbox",
@@ -213,6 +213,22 @@ export async function sandboxAcrossStepsWorkflow(
  return await inspectSandboxMarkerStep(ctx, marker);
 }

+export async function sharedSandboxWorkflow(_input: DeployInput, ctx: WorkflowToolContext) {
+  "use workflow";
+  await appendSharedSandboxStep(ctx);
+  await workflowSleep("10ms");
+  return await appendSharedSandboxStep(ctx);
+}
+
+async function appendSharedSandboxStep(ctx: WorkflowToolContext) {
+  "use step";
+  const sandbox = await ctx.getSandbox();
+  const previous = await sandbox.readTextFile({ path: "shared.txt" });
+  const content = `${previous}|workflow`;
+  await sandbox.writeTextFile({ path: "shared.txt", content });
+  return { id: sandbox.id, content };
+}
+
 export async function recoverSandboxFailureWorkflow(_input: DeployInput, ctx: WorkflowToolContext) {
  "use workflow";
  return await recoverSandboxFailureStep(ctx);
@@ -19,10 +19,10 @@ Use `defineWorkflowTool({ execute })` and pass `ctx` directly to a `"use step"`
 - A workflow that never calls `getSandbox` does not open a sandbox.
 - On first access in each step, the step requests initialization through its owner inbox. The owning session checks the recorded run, opens or reconnects its sandbox, and persists the updated session checkpoint before returning a serializable reconnect record. Background tasks forward this request through their existing parent delivery path.
 - A durable response stream keyed by the requesting step lets retries reuse the response. Concurrent steps are initialized through the owning session so they share its initialization state. This adds an owner round trip on first access in each step.
- The existing workflow step context wrapper reconnects the backend and binds operations to the step's abort signal. Repeated calls in one step reuse its handle.
+- Each workflow step binds a lazy `SandboxAccess` under `SandboxKey` and uses the existing tool getter and cancellation wrapper. Calls in one step share access; later steps reconstruct it from the session's saved state.
 - Steps cannot stop or delete the shared sandbox. They consume or kill spawned processes before returning and return serializable results, never live handles or streams.
 - Sandbox expiration and backend failures retain the backend's existing recovery behavior; this change adds no new persistence guarantees.

 ## Validation

-Runtime integration coverage checks file persistence across a durable sleep and successive steps for blocking and background tools, plus concurrent first accesses, step retries, and a clear failure from the workflow body. Fixture evals exercise the same contract through an agent. CI is required for the fixture evals.
+Runtime integration coverage checks regular tools and workflow steps reading and updating the same sandbox. It also covers durable waits, concurrent first accesses, step retries, and rejected access from the workflow body for blocking and background tools. Fixture evals exercise the same contract through an agent. CI is required for the fixture evals.