diff --git a/.changeset/sandbox-commands-tools.md b/.changeset/sandbox-commands-tools.md new file mode 100644 index 000000000..ccc16a442 --- /dev/null +++ b/.changeset/sandbox-commands-tools.md @@ -0,0 +1,21 @@ +--- +"inngest": patch +--- + +Experimental: code running in a sandbox can now call functions in your app. Pass `tools` to `commands.run`, and each call the code makes runs your function as its own durable step, so it's retried and memoized like any other step. + +```ts +await sandbox.commands.run("agent", "node agent.js", { + tools: { + getCustomer: (id: string) => db.customers.find(id), + }, +}); +``` + +```js +// agent.js, inside the sandbox +const { tools } = require("inngest_tools"); +const customer = await tools.getCustomer("cus_1"); +``` + +In Python, use `from inngest_tools import tools` and `await tools.getCustomer("cus_1")`. A tool that fails raises a `ToolError` the code can catch. diff --git a/packages/inngest/src/components/sandbox/durable.ts b/packages/inngest/src/components/sandbox/durable.ts index 79c29bcdc..cef37f721 100644 --- a/packages/inngest/src/components/sandbox/durable.ts +++ b/packages/inngest/src/components/sandbox/durable.ts @@ -26,6 +26,7 @@ import { type SandboxResultForAction, validateSandboxResult, } from "./protocol.ts"; +import type { SandboxToolsRequest } from "./tools.ts"; import { sdkOrigin } from "./trace.ts"; import { type DurableSandbox, @@ -52,6 +53,7 @@ import { decodeOutputChunk, decodeUtf8, encodeBase64, + maxSandboxProcessTimeoutMs, normalizeSandboxCommandOptions, normalizeSandboxCreateOptions, normalizeSandboxListOptions, @@ -480,9 +482,17 @@ const callRawTool = async < idOrOptions: StepOptionsOrId, operation: SandboxOperationForAction, finish?: (result: SandboxResultForAction) => T | Promise, + tools?: SandboxToolsRequest, ): Promise => { try { const rawTool = await rawToolResolver(); + if (tools) { + return validateSandboxResult( + operation, + await rawTool(idOrOptions, operation, undefined, tools), + ) as T; + } + if (finish) { return (await rawTool(idOrOptions, operation, (result) => { return finish(validateSandboxResult(operation, result)); @@ -755,7 +765,13 @@ export const createDurableSandboxFacade = ( resources: Object.freeze({ ...ref.resources }), commands: Object.freeze({ run: async (idOrOptions, command, options) => { - const normalized = normalizeSandboxCommandOptions(command, options); + const { tools, toolOptions, ...commandOptions } = options ?? {}; + const normalized = normalizeSandboxCommandOptions(command, { + ...commandOptions, + ...(tools && { + timeout: commandOptions.timeout ?? maxSandboxProcessTimeoutMs, + }), + }); const { timeout: _timeout, ...input } = normalized; const operation = parseSandboxOperationForAction("exec", { protocolVersion: sandboxProtocolVersion, @@ -767,6 +783,8 @@ export const createDurableSandboxFacade = ( rawToolResolver, idOrOptions, operation, + undefined, + tools && { tools, toolOptions }, ); return { stdout: decodeUtf8( diff --git a/packages/inngest/src/components/sandbox/middleware.ts b/packages/inngest/src/components/sandbox/middleware.ts index 50f5d9d52..cbb4a81bd 100644 --- a/packages/inngest/src/components/sandbox/middleware.ts +++ b/packages/inngest/src/components/sandbox/middleware.ts @@ -1,7 +1,10 @@ +import type { StepOptionsOrId } from "../../types.ts"; import { getAsyncCtxSync } from "../execution/als.ts"; import type { Inngest } from "../Inngest.ts"; +import type { MetadataKind, MetadataOpcode } from "../InngestMetadata.ts"; import { Middleware } from "../middleware/middleware.ts"; import { NonRetriableError } from "../NonRetriableError.ts"; +import { sandboxForOperation } from "./client.ts"; import { createSandboxTools, executeSandboxOperation } from "./durable.ts"; import { createGroupSandbox, type GroupSandbox } from "./group.ts"; import { @@ -12,7 +15,12 @@ import { type SandboxRawTool, } from "./protocol.ts"; import { createOneShotRunner, type OneShotRunner } from "./runners.ts"; -import { sandboxMetadataKind, sandboxTraceMetadata } from "./trace.ts"; +import { runCommandWithTools } from "./toolRunner.ts"; +import { + type SandboxTraceMetadata, + sandboxMetadataKind, + sandboxTraceMetadata, +} from "./trace.ts"; import { type DurableSandboxTools, SandboxError, @@ -32,12 +40,13 @@ type SandboxGroupExtension = { }; /** - * Attach `inngest.sandbox` metadata to the step that's executing, describing - * the sandbox action for the trace. Tracing must never fail the step. + * Attach metadata to the step that's executing. Tracing must never fail the + * step. */ -const describeStep = ( - operation: unknown, - outcome: { result: SandboxOperationResultV1 } | { error: unknown }, +const addStepMetadata = ( + kind: MetadataKind, + values: () => Record | undefined, + op: MetadataOpcode = "merge", ): void => { try { const execution = getAsyncCtxSync()?.execution; @@ -46,39 +55,86 @@ const describeStep = ( return; } - const parsed = parseSandboxOperation(operation); - - execution.instance.addMetadata( - stepId, - sandboxMetadataKind, - "step", - "merge", - sandboxTraceMetadata( - parsed, - "result" in outcome - ? outcome - : { error: getSandboxError(outcome.error) }, - ), + const metadata = values(); + + if (metadata) { + execution.instance.addMetadata(stepId, kind, "step", op, metadata); + } + } catch { + // An operation that fails validation has nothing to describe. + } +}; + +const describeStep = ( + operation: unknown, + outcome: { result: SandboxOperationResultV1 } | { error: unknown }, +): void => { + addStepMetadata(sandboxMetadataKind, () => { + return sandboxTraceMetadata( + parseSandboxOperation(operation), + "result" in outcome ? outcome : { error: getSandboxError(outcome.error) }, ); + }); + + if ("error" in outcome) { + addStepMetadata( + "inngest.warnings", + () => { + const parsed = parseSandboxOperation(operation); + + if (parsed.action !== "destroy") { + return; + } - if ("error" in outcome && parsed.action === "destroy") { - execution.instance.addMetadata( - stepId, - "inngest.warnings", - "step", - "set", - { + return { "sdk.sandbox.destroy": `Sandbox ${operationSandboxId(parsed)} couldn't be destroyed: ${ outcome.error instanceof Error ? outcome.error.message : String(outcome.error) }`, - }, - ); + }; + }, + "set", + ); + } +}; + +/** + * What a failed sandbox step throws: retryable errors retry, the rest do not. + */ +const toStepError = (error: unknown): unknown => { + if (error instanceof SandboxError) { + const cause = { + protocolVersion: error.protocolVersion, + action: error.action, + code: error.code, + message: error.message, + ...(error.status !== undefined && { status: error.status }), + ...(error.sandboxId !== undefined && { sandboxId: error.sandboxId }), + ...(error.processId !== undefined && { processId: error.processId }), + ...(error.snapshotId !== undefined && { snapshotId: error.snapshotId }), + ambiguous: error.ambiguous, + retryable: error.retryable, + ...(error.requestId !== undefined && { requestId: error.requestId }), + details: [...error.details], + }; + if (error.retryable) { + const retryableError = new Error(error.message, { cause }); + retryableError.name = error.name; + return retryableError; } - } catch { - // An operation that fails validation has nothing to describe. + return new NonRetriableError(error.message, { cause }); } + if (error instanceof SandboxValidationError) { + return new NonRetriableError(error.message, { + cause: { + protocolVersion: sandboxProtocolVersion, + type: "sandbox_validation_error" as const, + message: error.message, + }, + }); + } + return error; }; const executeAsStep = async ( @@ -91,46 +147,7 @@ const executeAsStep = async ( return result; } catch (error) { describeStep(operation, { error }); - if (error instanceof SandboxError) { - const cause = { - protocolVersion: error.protocolVersion, - action: error.action, - code: error.code, - message: error.message, - ...(error.status !== undefined && { status: error.status }), - ...(error.sandboxId !== undefined && { - sandboxId: error.sandboxId, - }), - ...(error.processId !== undefined && { - processId: error.processId, - }), - ...(error.snapshotId !== undefined && { - snapshotId: error.snapshotId, - }), - ambiguous: error.ambiguous, - retryable: error.retryable, - ...(error.requestId !== undefined && { - requestId: error.requestId, - }), - details: [...error.details], - }; - if (error.retryable) { - const retryableError = new Error(error.message, { cause }); - retryableError.name = error.name; - throw retryableError; - } - throw new NonRetriableError(error.message, { cause }); - } - if (error instanceof SandboxValidationError) { - throw new NonRetriableError(error.message, { - cause: { - protocolVersion: sandboxProtocolVersion, - type: "sandbox_validation_error" as const, - message: error.message, - }, - }); - } - throw error; + throw toStepError(error); } }; @@ -154,8 +171,52 @@ export class SandboxMiddleware extends Middleware.BaseMiddleware { SandboxGroupExtension; }; } { - const rawTool: SandboxRawTool = (idOrOptions, operation, finish) => - arg.ctx.step.run( + const { step, runId } = arg.ctx; + const rawTool: SandboxRawTool = (idOrOptions, operation, finish, tools) => { + if (tools && operation.action === "exec") { + return runCommandWithTools( + { + run: async ( + idOrOptions: StepOptionsOrId, + fn: () => Promise, + meta?: SandboxTraceMetadata, + ) => { + return (await step.run(idOrOptions, async () => { + try { + const result = await fn(); + if (meta) { + addStepMetadata(sandboxMetadataKind, () => { + return meta; + }); + } + + return result; + } catch (error) { + if (meta) { + addStepMetadata(sandboxMetadataKind, () => { + const code = getSandboxError(error)?.code; + + return code ? { ...meta, error_code: code } : meta; + }); + } + + throw toStepError(error); + } + })) as T; + }, + sandbox: sandboxForOperation( + this.client.sandboxes, + operation.target.sandbox, + ), + runId, + }, + idOrOptions, + operation, + tools, + ); + } + + return step.run( idOrOptions, async (input) => { const result = await executeAsStep(this.client, input); @@ -164,6 +225,7 @@ export class SandboxMiddleware extends Middleware.BaseMiddleware { }, operation, ); + }; const sandbox = createSandboxTools(() => rawTool); diff --git a/packages/inngest/src/components/sandbox/protocol.ts b/packages/inngest/src/components/sandbox/protocol.ts index 71de8edf3..41af3a938 100644 --- a/packages/inngest/src/components/sandbox/protocol.ts +++ b/packages/inngest/src/components/sandbox/protocol.ts @@ -1,6 +1,7 @@ import { z } from "zod/v3"; import type { StepOptionsOrId } from "../../types.ts"; +import type { SandboxToolsRequest } from "./tools.ts"; import { type SandboxAction, SandboxError, @@ -675,6 +676,10 @@ export type SandboxRawTool = ( idOrOptions: StepOptionsOrId, operation: SandboxOperationV1, finish?: (result: unknown) => unknown, + /** + * Makes an `exec` run with tools the code can call. + */ + tools?: SandboxToolsRequest, ) => Promise; export type SandboxRawToolResolver = () => | SandboxRawTool diff --git a/packages/inngest/src/components/sandbox/toolRunner.test.ts b/packages/inngest/src/components/sandbox/toolRunner.test.ts index df989ef29..606f1cb20 100644 --- a/packages/inngest/src/components/sandbox/toolRunner.test.ts +++ b/packages/inngest/src/components/sandbox/toolRunner.test.ts @@ -8,7 +8,12 @@ import type { StepOptions } from "../../types.ts"; import { getAsyncCtxSync, runWithAsyncCtx } from "../execution/als.ts"; import { withSpan } from "../InngestStepTools.ts"; import { NonRetriableError } from "../NonRetriableError.ts"; -import { parseSandboxOperationForAction } from "./protocol.ts"; +import { createDurableSandboxFacade } from "./durable.ts"; +import { + parseSandboxOperationForAction, + type SandboxRawTool, +} from "./protocol.ts"; +import { createClient, runToEnd, sandboxResource } from "./test-helpers.ts"; import { runCommandWithTools, type ToolsHost } from "./toolRunner.ts"; import type { SandboxTools } from "./tools.ts"; import { sdkOrigin } from "./trace.ts"; @@ -713,6 +718,40 @@ describe("commands.run with tools", () => { expect.stringMatching(/^\/inngest\/tools\/[0-9a-f]{64}$/), ]); }); + + test("a repeated command ID gets its own directory", async () => { + const dirs = new Set(); + + for (const stepId of ["hashed", "hashed:1"]) { + const { fake, sandbox } = createFakeSandbox((f, replied) => { + if (replied === undefined) { + exitWith(f); + } + }); + const { host } = createHost(sandbox); + const inStep: ToolsHost["run"] = (idOrOptions, fn, meta) => { + return host.run( + idOrOptions, + () => { + return runWithAsyncCtx( + { + app: {} as never, + execution: { executingStep: { id: stepId } } as never, + }, + fn, + ); + }, + meta, + ); + }; + + await run({ ...host, run: inStep }, {}); + + dirs.add(fake.processCommand?.[4]); + } + + expect(dirs.size).toBe(2); + }); }); describe("commands.run with tools trace group", () => { @@ -827,3 +866,113 @@ describe("commands.run with tools trace group", () => { ]); }); }); + +describe("durable commands.run routing", () => { + const execResult = { + protocolVersion: 1, + action: "exec", + result: { + stdout: "", + stderr: "", + encoding: "base64", + exitCode: 0, + output: { truncated: false }, + }, + }; + + test.each([ + { name: "without tools", options: undefined, tools: undefined }, + { + name: "with tools, and a 5m default timeout", + options: { tools: { a: async () => 1 } }, + tools: { tools: { a: expect.any(Function) }, toolOptions: undefined }, + }, + ])("$name it is one raw tool call", async ({ options, tools }) => { + const rawTool: SandboxRawTool = vi.fn(async () => { + return execResult; + }); + const sandbox = createDurableSandboxFacade(sandboxRef, () => { + return rawTool; + }); + + await sandbox.commands.run("exec", "true", options); + + expect(rawTool).toHaveBeenCalledExactlyOnceWith( + "exec", + expect.objectContaining({ + input: [ + expect.objectContaining({ timeoutMs: tools ? 300_000 : 30_000 }), + ], + }), + ...(tools ? [undefined, tools] : []), + ); + }); +}); + +describe("commands.run with tools in a function", () => { + test("starts the command, follows it and returns its output", async () => { + const wrapped: string[][] = []; + const exit = JSON.stringify({ + op: "exit", + code: 0, + stdout: base64("hello\n"), + stderr: "", + }); + const client = createClient((path, method, { body }) => { + if (path === "/v2/sandboxes" && method === "POST") { + return Response.json({ data: sandboxResource }, { status: 201 }); + } + + if (path.endsWith("/processes") && method === "POST") { + wrapped.push((body ? JSON.parse(String(body)) : {}).command); + + return Response.json( + { + data: { + id: "33333333-3333-4333-8333-333333333333", + command: ["sh"], + pid: 1, + state: "RUNNING", + }, + }, + { status: 201 }, + ); + } + + if (path.endsWith("/exec")) { + return Response.json({ + data: { + stdout: base64(`${exit}\n`), + stderr: "", + encoding: "base64", + exitCode: 0, + }, + }); + } + + return undefined; + }); + let stdout = ""; + const fn = client.createFunction( + { id: "tools", triggers: [{ event: "tools/run" }] }, + async ({ step }) => { + const sandbox = await step.sandbox.create("box", { + name: "box", + vcpu: 1, + memoryMb: 512, + }); + const result = await sandbox.commands.run("hello", "echo hello", { + tools: { a: async () => 1 }, + }); + + stdout = result.stdout; + }, + ); + + const { ran } = await runToEnd(fn); + + expect(ran).toEqual(["box", "hello"]); + expect(wrapped).toHaveLength(1); + expect(stdout).toBe("hello\n"); + }); +}); diff --git a/packages/inngest/src/components/sandbox/toolRunner.ts b/packages/inngest/src/components/sandbox/toolRunner.ts index 4d5a3bfbf..1e4119e07 100644 --- a/packages/inngest/src/components/sandbox/toolRunner.ts +++ b/packages/inngest/src/components/sandbox/toolRunner.ts @@ -185,9 +185,6 @@ const followCommand = async ( const [{ command, environment, cwd, timeoutMs }] = operation.input; const { sandbox } = host; - const dir = `${toolsRoot}/${sha256() - .update(`${host.runId ?? ""}:${id}`) - .digest("hex")}`; const meta = sandboxTraceMetadata(operation, { error: undefined }); const sdkStep = (stepId: string, name: string): StepOptions => { @@ -198,7 +195,7 @@ const followCommand = async ( * Replies ride in the waiter's command while they're small. A larger one is * uploaded, and the command names its file. */ - const stage = (replies: string[]): Promise => { + const stage = (dir: string, replies: string[]): Promise => { let inlineBytes = 0; return Promise.all( @@ -226,6 +223,7 @@ const followCommand = async ( * ask for anything new until it has its reply. */ const follow = async ( + dir: string, lines: number, replies: string[], windowMs: number, @@ -240,7 +238,7 @@ const followCommand = async ( dir, `${lines}`, `${seconds}`, - ...(await stage(replies)), + ...(await stage(dir, replies)), ], { timeout: (seconds + 30) * 1000 }, ); @@ -265,7 +263,7 @@ const followCommand = async ( return { now: Date.now(), lines: lines + log.length, requests, result }; }; - const abort = async (reason: string): Promise => { + const abort = async (dir: string, reason: string): Promise => { await host.run(sdkStep(`${id}:abort`, "Stop command"), async () => { await sandbox.commands.run([ "/bin/sh", @@ -326,6 +324,11 @@ const followCommand = async ( idOrOptions, async () => { const startedAt = Date.now(); + // The step's own ID, which stays distinct when a command ID repeats. + const stepId = getAsyncCtxSync()?.execution?.executingStep?.id ?? id; + const dir = `${toolsRoot}/${sha256() + .update(`${host.runId ?? ""}:${stepId}`) + .digest("hex")}`; // The wait doesn't need the process to exist yet, so it runs alongside. const [, round] = await Promise.all([ sandbox.processes.start({ @@ -333,14 +336,15 @@ const followCommand = async ( ...(environment && { environment }), ...(cwd !== undefined && { cwd }), }), - follow(0, [], Math.min(waitWindowMs, timeoutMs)), + follow(dir, 0, [], Math.min(waitWindowMs, timeoutMs)), ]); - return { ...round, startedAt }; + return { ...round, startedAt, dir }; }, meta, ); + const { dir } = started; const seen = new Set(); let state: Round = started; let waits = 0; @@ -351,7 +355,10 @@ const followCommand = async ( } if (state.now - started.startedAt > timeoutMs) { - return abort(`Command with tools exceeded its ${timeoutMs}ms timeout`); + return abort( + dir, + `Command with tools exceeded its ${timeoutMs}ms timeout`, + ); } const pending: Round["requests"] = []; @@ -363,7 +370,7 @@ const followCommand = async ( seen.add(request.id); if (seen.size > maxCalls) { - return abort(`Command made more than ${maxCalls} tool calls`); + return abort(dir, `Command made more than ${maxCalls} tool calls`); } pending.push({ request, line }); @@ -383,7 +390,7 @@ const followCommand = async ( state = await host.run( sdkStep(`${id}:wait:${++waits}`, runningNames[commandKind(command)]), () => { - return follow(lines, replies, Math.min(waitWindowMs, remainingMs)); + return follow(dir, lines, replies, Math.min(waitWindowMs, remainingMs)); }, { ...meta, kind: "wait" }, ); diff --git a/packages/inngest/src/components/sandbox/types.ts b/packages/inngest/src/components/sandbox/types.ts index d1003c658..42945b5ea 100644 --- a/packages/inngest/src/components/sandbox/types.ts +++ b/packages/inngest/src/components/sandbox/types.ts @@ -1,5 +1,6 @@ import type { DurationLike } from "../../helpers/temporal.ts"; import type { StepOptionsOrId } from "../../types.ts"; +import type { SandboxToolsOptions } from "./tools.ts"; export const sandboxProtocolVersion = 1 as const; @@ -561,7 +562,7 @@ export interface DurableSandbox { run( idOrOptions: StepOptionsOrId, command: SandboxCommand, - options?: SandboxCommandOptions, + options?: SandboxCommandOptions & SandboxToolsOptions, ): Promise; }; /** diff --git a/packages/inngest/src/experimental.ts b/packages/inngest/src/experimental.ts index 84a2d3559..a8b23dad1 100644 --- a/packages/inngest/src/experimental.ts +++ b/packages/inngest/src/experimental.ts @@ -82,4 +82,9 @@ export { // Scoring export { scoreMiddleware } from "./components/InngestScore.ts"; export { createScorer } from "./components/ScoreFunction.ts"; +export type { + SandboxToolOptions, + SandboxTools, + SandboxToolsOptions, +} from "./components/sandbox/tools.ts"; export type { ExperimentRef } from "./types.ts";