diff --git a/.changeset/exec-metadata.md b/.changeset/exec-metadata.md new file mode 100644 index 00000000..7aea7918 --- /dev/null +++ b/.changeset/exec-metadata.md @@ -0,0 +1,5 @@ +--- +"@cloudflare/computer": minor +--- + +`runtime.exec()` takes `metadata`, a record of strings for host modules that the run's code never sees. Host module calls read it as `context.execution.metadata`, `runtime.pending()` lists it with each waiting call, and a replayable run keeps it when it resumes. `WorkerJavaScriptBackend` caps it at 16 KiB of keys and values. diff --git a/docs/05_runtime_interface.md b/docs/05_runtime_interface.md index e32aad85..f6bcfa02 100644 --- a/docs/05_runtime_interface.md +++ b/docs/05_runtime_interface.md @@ -33,6 +33,7 @@ interface WorkspaceRuntimeExecOptions { timeoutMs?: number; env?: Record; stdin?: Uint8Array | string; + metadata?: Record; output?: { maxLines?: number; maxBytes?: number } | false; } @@ -45,7 +46,7 @@ interface WorkspaceRuntimeExecHandle extends ReadableStream { + const { conversation } = execution.metadata; + if (conversation === undefined) throw new Error("This run isn't for a conversation."); + return notes.add(conversation, String(text)); + }, + }, +} +``` + +Host code reads metadata as `WorkspaceExecutionMetadata`, whose values may be `undefined`, because any key may be missing: a run may have been started without metadata or by other code, and a run saved before metadata existed resumes with an empty object. + +The run's code never sees metadata: it isn't in `process.env` or the input, so code can't read or change it, and a module can trust it. It's capped at 16 KiB of keys and values. A replayable run keeps its metadata when it resumes, `pending()` lists it with each waiting call, and `workspace.stub().runtime.exec()` takes it too. Backends that run no host modules ignore it. + Specifiers and the export names of an object are checked at construction. A factory's export names are checked when the backend connects and the factory runs. An object must export at least one function. A factory may build none, as `codemode:tools` does for an agent with no tools: the backend then leaves that module out of the connection, so only code that imports it fails, with a message saying the module exports nothing, and every other run still starts. Every export name must be a JavaScript identifier name other than `default` or `then`. A reserved word such as `delete` is allowed, and caller code renames it on import: `import { delete as remove } from "codemode:files"`. Importing a name the module does not export fails when the module graph links, before any code runs. ### `codemode:git` diff --git a/examples/rlm/worker/capability.test.ts b/examples/rlm/worker/capability.test.ts index fbf1b6c7..00905a4f 100644 --- a/examples/rlm/worker/capability.test.ts +++ b/examples/rlm/worker/capability.test.ts @@ -40,7 +40,7 @@ function context(signal = new AbortController().signal): WorkspaceModuleCallCont deadline: Date.now() + 1_000, access: "read", callId: "test-call", - execution: { id: "test-run" }, + execution: { id: "test-run", metadata: {} }, interrupt: () => Promise.reject(new Error("not used")), resolvePath: async (path) => path, }; diff --git a/packages/computer/src/backends/worker-javascript/journal.ts b/packages/computer/src/backends/worker-javascript/journal.ts index 3d6211b5..810a08e1 100644 --- a/packages/computer/src/backends/worker-javascript/journal.ts +++ b/packages/computer/src/backends/worker-javascript/journal.ts @@ -72,6 +72,12 @@ export interface ResumeStart { readonly seed: string; /** Identifies the runtime the execution started on. */ readonly fingerprint: string; + /** + * The run's metadata for host modules. Runs saved before metadata + * existed have none and resume with an empty object, which is why + * adding it left `JOURNAL_VERSION` as it was. + */ + readonly metadata?: Record; } export interface ResumeState { diff --git a/packages/computer/src/backends/worker-javascript/worker-javascript.test.ts b/packages/computer/src/backends/worker-javascript/worker-javascript.test.ts index e0797f23..499f045a 100644 --- a/packages/computer/src/backends/worker-javascript/worker-javascript.test.ts +++ b/packages/computer/src/backends/worker-javascript/worker-javascript.test.ts @@ -4,7 +4,7 @@ import { describe, expect, it, vi } from "vitest"; import type { WorkspaceRuntimeLoader } from "../../runtime/types"; import { Workspace } from "../../workspace"; -import { ExecutionJournal, fingerprint, loadResume } from "./journal"; +import { ExecutionJournal, fingerprint, loadResume, type ResumeStart, saveResume } from "./journal"; import { runtimeCode, WorkerJavaScriptBackend } from "./worker-javascript"; function throwingLoader(message: string) { @@ -377,6 +377,39 @@ describe("WorkerJavaScriptBackend", () => { expect(load).not.toHaveBeenCalled(); }); + it("rejects metadata larger than 16 KiB", async () => { + const load = vi.fn(); + const workspace = new Workspace({ + storage: new SQLiteTestStorage(), + backends: [new WorkerJavaScriptBackend({ loader: { load } })], + }); + await workspace.fs.mkdir("/workspace", { recursive: true }); + await expect( + workspace.runtime.exec("export default 1", { metadata: { key: "x".repeat(16 * 1024) } }), + ).rejects.toThrow(/metadata exceeds 16384 bytes/); + expect(load).not.toHaveBeenCalled(); + }); + + it("rejects metadata that isn't a string-to-string record", async () => { + const load = vi.fn(); + const workspace = new Workspace({ + storage: new SQLiteTestStorage(), + backends: [new WorkerJavaScriptBackend({ loader: { load } })], + }); + await workspace.fs.mkdir("/workspace", { recursive: true }); + await expect( + workspace.runtime.exec("export default 1", { + metadata: { key: 42 as unknown as string }, + }), + ).rejects.toThrow(/metadata value for "key" must be a string/); + await expect( + workspace.runtime.exec("export default 1", { + metadata: ["a"] as unknown as Record, + }), + ).rejects.toThrow(/metadata must be a string-to-string record/); + expect(load).not.toHaveBeenCalled(); + }); + it("checks limits against the complete loader map including the runtime runner", async () => { const load = vi.fn(); const workspace = new Workspace({ @@ -2040,7 +2073,11 @@ describe("WorkerJavaScriptBackend interrupted executions", () => { }); } - async function interrupt(storage: SQLiteTestStorage, timeoutMs?: number) { + async function interrupt( + storage: SQLiteTestStorage, + timeoutMs?: number, + metadata?: Record, + ) { const workspace = open(storage); await workspace.fs.mkdir("/workspace", { recursive: true }); await workspace.runtime.exec("export default 1", { @@ -2048,10 +2085,53 @@ describe("WorkerJavaScriptBackend interrupted executions", () => { id: "run", onInterrupt: "replay", ...(timeoutMs === undefined ? {} : { timeoutMs }), + ...(metadata === undefined ? {} : { metadata }), }); return workspace; } + // Rewrite the run's saved start as a version without metadata wrote it. + function forgetMetadata(storage: SQLiteTestStorage) { + const db = new Database(storage); + const saved = loadResume(db, "worker-javascript", "run"); + if (saved === undefined) throw new Error("expected a saved run"); + const { metadata: _metadata, ...start } = saved.start; + saveResume(db, "worker-javascript", "run", saved.policy, start as ResumeStart); + } + + // A loader whose runs call codemode:test.who once and then hang, so the run + // stays running. + function calling() { + return { + load() { + return { + getEntrypoint() { + return { + async evaluate( + _input: unknown, + host: { call(name: string, args: unknown[]): Promise }, + ) { + await host.call("host/codemode:test.who", []); + await new Promise(() => undefined); + }, + }; + }, + }; + }, + }; + } + + function recordingModule(seen: unknown[]) { + return { + "codemode:test": { + who: (_args: unknown, context: { execution: unknown }) => { + seen.push(context.execution); + return null; + }, + }, + }; + } + async function settle(workspace: Workspace) { const handle = await workspace.runtime.getExec("run", { backend: "worker-javascript", @@ -2182,6 +2262,53 @@ describe("WorkerJavaScriptBackend interrupted executions", () => { for (const workspace of workspaces) await workspace.close(); }); + it("keeps a run's metadata when it resumes", async () => { + const storage = new SQLiteTestStorage(); + const first = await interrupt(storage, undefined, { conversation: "c1" }); + const seen: unknown[] = []; + const later = open(storage, { loader: calling(), modules: recordingModule(seen) }); + await later.runtime.recover(); + await vi.waitFor(() => expect(seen).toHaveLength(1)); + expect(seen).toEqual([{ id: "run", metadata: { conversation: "c1" } }]); + await later.close(); + await first.close(); + }); + + it("resumes a run saved without metadata, with none", async () => { + const storage = new SQLiteTestStorage(); + const first = await interrupt(storage); + forgetMetadata(storage); + const seen: unknown[] = []; + const later = open(storage, { loader: calling(), modules: recordingModule(seen) }); + await later.runtime.recover(); + await vi.waitFor(() => expect(seen).toHaveLength(1)); + expect(seen).toEqual([{ id: "run", metadata: {} }]); + await later.close(); + await first.close(); + }); + + it("lists a waiting run's metadata, and none for a run saved without it", async () => { + for (const [metadata, listed] of [ + [{ conversation: "c1" }, { conversation: "c1" }], + [undefined, {}], + ] as const) { + const storage = new SQLiteTestStorage(); + const db = new Database(storage); + const first = await interrupt(storage, undefined, metadata); + if (metadata === undefined) forgetMetadata(storage); + const journal = new ExecutionJournal(db, "worker-javascript", "run"); + const step = journal.begin("host/codemode:test.approve", []); + if (step.kind !== "live") throw new Error("expected a live call"); + journal.interrupt(step.seq, { kind: "approval" }); + const later = open(storage); + await expect(later.runtime.pending()).resolves.toEqual([ + expect.objectContaining({ id: "run", metadata: listed }), + ]); + await later.close(); + await first.close(); + } + }); + it("asks for a REPLAY_VERSION check when the runtime's code changes", () => { // The code the runtime loads into the isolate changed. If a run that // started on the old code could replay differently on the new code, diff --git a/packages/computer/src/backends/worker-javascript/worker-javascript.ts b/packages/computer/src/backends/worker-javascript/worker-javascript.ts index 42c7ce08..a2a79f8c 100644 --- a/packages/computer/src/backends/worker-javascript/worker-javascript.ts +++ b/packages/computer/src/backends/worker-javascript/worker-javascript.ts @@ -9,10 +9,12 @@ import { import type { ModuleExecutionEnvelope, ModuleExecutionInput, + WorkspaceExecutionMetadata, WorkspaceModule, WorkspaceModuleBackend, WorkspaceModuleBackendHandle, WorkspaceModuleBackendHost, + WorkspaceModuleExecution, WorkspaceModuleFunctions, WorkspaceRuntimeAccess, WorkspaceRuntimeEvent, @@ -259,6 +261,11 @@ interface ExecutionSubscriber { interface ExecutionRecord { id: string; + // What host module calls are told about the run. Set while the run + // is in memory, from `exec()` or, for a run resumed after a restart, + // from its saved start. Absent on a record loaded only to replay its + // output. + execution?: WorkspaceModuleExecution; events: WorkspaceRuntimeEvent[]; subscribers: Set; status: "running" | "completed" | "failed" | "cancelled"; @@ -511,6 +518,7 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { this.#records.set(id, record); const saved = this.#loadResume(record); if (saved === undefined) return; + record.execution = this.#execution(id, saved.start.metadata); // A run waiting for an answer resumes once it has every one. if (saved.waitingSince !== null) { record.waiting = "released"; @@ -611,6 +619,7 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { callId: `${record.id}:${call.seq}`, payload: call.payload as WorkspaceRuntimeValue, since: call.since, + metadata: this.#metadataOf(record.id), }; // After the current call, so the hook can't hold up the run's release. void Promise.resolve() @@ -634,9 +643,33 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { callId: `${call.id}:${call.seq}`, payload: call.payload as WorkspaceRuntimeValue, since: call.since, + metadata: this.#metadataOf(call.id), })); } + // A waiting run's metadata. Every running record is in memory with its + // execution, so the saved start is only read for a run that isn't. + #metadataOf(id: string): WorkspaceExecutionMetadata { + const known = this.#records.get(id)?.execution; + if (known !== undefined) return known.metadata; + let saved: ReturnType; + try { + saved = loadResume(this.#host.db, this.#backendId, id); + } catch { + saved = undefined; + } + return this.#execution(id, saved?.start.metadata).metadata; + } + + // What host module calls of a run are told about it, frozen so one + // module can't change what another sees. + #execution(id: string, metadata: Record | undefined): WorkspaceModuleExecution { + return Object.freeze({ + id, + metadata: Object.freeze({ ...metadata }), + }); + } + async respond(input: { id: string; callId: string; answer: WorkspaceRuntimeValue }) { assertExecutionId(input.id); assertRuntimeValue(input.answer); @@ -808,7 +841,8 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { if (stdinBytes.byteLength > this.#options.maxStdinBytes) { throw new Error(`Workspace runtime stdin exceeds ${this.#options.maxStdinBytes} bytes.`); } - assertEnv(input.env, this.#options.maxEnvBytes); + assertStringRecord(input.env, this.#options.maxEnvBytes, "env"); + assertStringRecord(input.metadata, MAX_METADATA_BYTES, "metadata"); const policy = input.onInterrupt ?? this.#options.onInterrupt; assertInterruptPolicy(policy); if (new TextEncoder().encode(input.source).byteLength > this.#options.maxSourceBytes) { @@ -839,6 +873,7 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { } const record: ExecutionRecord = { id, + execution: this.#execution(id, input.metadata), events: [], subscribers: new Set(), status: "running", @@ -887,6 +922,7 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { startedAt, seed, fingerprint: this.#fingerprint(), + metadata: { ...input.metadata }, }); if (policy === "replay") { journal = this.#journalFor(record); @@ -944,7 +980,7 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { maxTotalResponseBytes: this.#options.maxCapabilityResponseBytes, maxResultBytes: this.#options.maxResultBytes, onAttachOutput: (readable) => this.#pumpFrames(record, readable), - execution: { id: record.id }, + execution: record.execution ?? this.#execution(record.id, undefined), onWait: (call) => { this.#notifyWait(record, call); void this.#release(record); @@ -1940,21 +1976,31 @@ function normalizeStdin(stdin: Uint8Array | string | undefined): Uint8Array { throw new Error("Workspace runtime stdin must be a string or Uint8Array."); } -function assertEnv(env: Record | undefined, maxBytes: number): void { - if (env === undefined) return; - if (typeof env !== "object" || env === null || Array.isArray(env)) { - throw new Error("Workspace runtime env must be a string-to-string record."); +// The most bytes of `metadata` one execution can carry, counting the +// UTF-8 bytes of every key and value. +const MAX_METADATA_BYTES = 16 * 1024; + +function assertStringRecord( + record: Record | undefined, + maxBytes: number, + name: "env" | "metadata", +): void { + if (record === undefined) return; + if (typeof record !== "object" || record === null || Array.isArray(record)) { + throw new Error(`Workspace runtime ${name} must be a string-to-string record.`); } let bytes = 0; const encoder = new TextEncoder(); - for (const [key, value] of Object.entries(env)) { + for (const [key, value] of Object.entries(record)) { if (typeof value !== "string") { - throw new Error(`Workspace runtime env value for ${JSON.stringify(key)} must be a string.`); + throw new Error( + `Workspace runtime ${name} value for ${JSON.stringify(key)} must be a string.`, + ); } bytes += encoder.encode(key).byteLength + encoder.encode(value).byteLength; } if (bytes > maxBytes) { - throw new Error(`Workspace runtime env exceeds ${maxBytes} bytes.`); + throw new Error(`Workspace runtime ${name} exceeds ${maxBytes} bytes.`); } } diff --git a/packages/computer/src/client.test.ts b/packages/computer/src/client.test.ts index f2cf73c6..98e08e79 100644 --- a/packages/computer/src/client.test.ts +++ b/packages/computer/src/client.test.ts @@ -360,10 +360,19 @@ describe("client runtime.exec — plain string form", () => { it("forwards options unchanged", async () => { const { host, calls } = fakeRemote(); const ws = await getWorkspace(host); - await ws.runtime.exec("npm test", { cwd: "/workspace", backend: "sandbox" }); + await ws.runtime.exec("npm test", { + cwd: "/workspace", + backend: "sandbox", + metadata: { conversation: "c1" }, + }); expect(calls[0]).toEqual({ command: "npm test", - options: { cwd: "/workspace", backend: "sandbox", id: expect.any(String) }, + options: { + cwd: "/workspace", + backend: "sandbox", + metadata: { conversation: "c1" }, + id: expect.any(String), + }, }); }); diff --git a/packages/computer/src/client.ts b/packages/computer/src/client.ts index 8e8b00d0..b1765df0 100644 --- a/packages/computer/src/client.ts +++ b/packages/computer/src/client.ts @@ -228,6 +228,8 @@ export interface RuntimeExecOptions { input?: WorkspaceRuntimeValue; env?: Record; stdin?: Uint8Array | string; + /** Strings for host modules, hidden from the run's code; see `WorkspaceRuntimeExecOptions.metadata`. */ + metadata?: Record; /** Output limits for this run; see `WorkspaceRuntimeExecOptions.output`. */ output?: WorkspaceRuntimeOutputOptions | false; } diff --git a/packages/computer/src/index.ts b/packages/computer/src/index.ts index 5901aaca..9e475d71 100644 --- a/packages/computer/src/index.ts +++ b/packages/computer/src/index.ts @@ -90,6 +90,7 @@ export type { WorkspaceBackendInfo } from "./runtime/runtime"; export type { ModuleExecutionEnvelope, ModuleExecutionInput, + WorkspaceExecutionMetadata, WorkspaceModule, WorkspaceModuleBackend, WorkspaceModuleBackendHandle, diff --git a/packages/computer/src/modules/artifacts.test.ts b/packages/computer/src/modules/artifacts.test.ts index 8e2787a9..22c1967c 100644 --- a/packages/computer/src/modules/artifacts.test.ts +++ b/packages/computer/src/modules/artifacts.test.ts @@ -44,7 +44,7 @@ function callContext(access: "read" | "read-write" = "read-write"): WorkspaceMod deadline: Date.now() + 60_000, access, callId: "test-call", - execution: { id: "test-run" }, + execution: { id: "test-run", metadata: {} }, interrupt: () => Promise.reject(new Error("not used")), resolvePath: async (path) => path, }; diff --git a/packages/computer/src/modules/assets.test.ts b/packages/computer/src/modules/assets.test.ts index f0079b6a..af953e0a 100644 --- a/packages/computer/src/modules/assets.test.ts +++ b/packages/computer/src/modules/assets.test.ts @@ -41,7 +41,7 @@ function callContext( deadline: Date.now() + 60_000, access: "read-write", callId: "test-call", - execution: { id: "test-run" }, + execution: { id: "test-run", metadata: {} }, interrupt: () => Promise.reject(new Error("not used")), resolvePath: async (path) => (path.startsWith("/") ? path : `/workspace/${path}`), ...overrides, diff --git a/packages/computer/src/modules/container.test.ts b/packages/computer/src/modules/container.test.ts index 221dbc9e..6c3bab61 100644 --- a/packages/computer/src/modules/container.test.ts +++ b/packages/computer/src/modules/container.test.ts @@ -90,7 +90,7 @@ function callContext( deadline: Date.now() + 60_000, access: "read-write", callId: "test-call", - execution: { id: "test-run" }, + execution: { id: "test-run", metadata: {} }, interrupt: () => Promise.reject(new Error("not used")), resolvePath: async (path) => path, ...overrides, diff --git a/packages/computer/src/modules/tools.test.ts b/packages/computer/src/modules/tools.test.ts index b02dfb98..d6dc1a73 100644 --- a/packages/computer/src/modules/tools.test.ts +++ b/packages/computer/src/modules/tools.test.ts @@ -18,7 +18,7 @@ function callContext( deadline: Date.now() + 60_000, access: "read-write", callId: "test-call", - execution: { id: "test-run" }, + execution: { id: "test-run", metadata: {} }, interrupt: () => Promise.reject(new Error("not used")), resolvePath: async (path) => path, ...overrides, diff --git a/packages/computer/src/runtime/bridge.test.ts b/packages/computer/src/runtime/bridge.test.ts index 89e6f023..e24548b0 100644 --- a/packages/computer/src/runtime/bridge.test.ts +++ b/packages/computer/src/runtime/bridge.test.ts @@ -313,7 +313,7 @@ describe("WorkspaceRuntimeBridge journal", () => { ], ]), journal, - execution: { id: "exec-1" }, + execution: { id: "exec-1", metadata: { conversation: "c1" } }, }); return { target, commits, responses, diverged, contexts, logs }; } @@ -362,11 +362,10 @@ describe("WorkspaceRuntimeBridge journal", () => { ]); await target.call("host/codemode:test.run", args); await target.call("host/codemode:test.run", args); - expect(contexts.map((context) => context.execution)).toEqual([ - { id: "exec-1" }, - { id: "exec-1" }, - ]); + const execution = { id: "exec-1", metadata: { conversation: "c1" } }; + expect(contexts.map((context) => context.execution)).toEqual([execution, execution]); expect(Object.isFrozen(contexts[0]?.execution)).toBe(true); + expect(Object.isFrozen(contexts[0]?.execution.metadata)).toBe(true); }); it("refuses to run an unsafe call again after an interruption", async () => { diff --git a/packages/computer/src/runtime/bridge.ts b/packages/computer/src/runtime/bridge.ts index 9603da9a..febcb2bc 100644 --- a/packages/computer/src/runtime/bridge.ts +++ b/packages/computer/src/runtime/bridge.ts @@ -80,7 +80,11 @@ export class WorkspaceRuntimeBridge extends RpcTarget { this.#onAttachOutput = integrations.onAttachOutput; this.#journal = integrations.journal; this.#fetch = integrations.fetch; - this.#execution = Object.freeze({ id: integrations.execution?.id ?? crypto.randomUUID() }); + const execution = integrations.execution; + this.#execution = Object.freeze({ + id: execution?.id ?? crypto.randomUUID(), + metadata: Object.freeze({ ...execution?.metadata }), + }); this.#onWait = integrations.onWait; } diff --git a/packages/computer/src/runtime/runtime.ts b/packages/computer/src/runtime/runtime.ts index 6d2db2bc..fde5136c 100644 --- a/packages/computer/src/runtime/runtime.ts +++ b/packages/computer/src/runtime/runtime.ts @@ -112,6 +112,7 @@ export class WorkspaceRuntime { timeoutMs: options.timeoutMs, sync: options.sync, onInterrupt: options.onInterrupt, + metadata: options.metadata, }); return wrapModuleHandle( runtime, diff --git a/packages/computer/src/runtime/types.ts b/packages/computer/src/runtime/types.ts index fdb0f77f..5a725ba7 100644 --- a/packages/computer/src/runtime/types.ts +++ b/packages/computer/src/runtime/types.ts @@ -14,8 +14,21 @@ export interface WorkspaceModuleExecution { * keeps its id when it resumes after an interruption or an answer. */ readonly id: string; + /** + * What the app passed as `metadata` to `exec()`, or an empty object. + * The run's own code never sees it, so a module can trust it to say + * which conversation or user the run is for. + */ + readonly metadata: WorkspaceExecutionMetadata; } +/** + * A run's metadata as host code reads it. Any key may be missing: the + * run may have been started without metadata, by other code, or before + * metadata existed, and then resumed with none. + */ +export type WorkspaceExecutionMetadata = Readonly>; + /** Per-call context the backend passes to every host module function. */ export interface WorkspaceModuleCallContext { /** Aborts when the call passes its deadline or the execution is cancelled. */ @@ -248,6 +261,15 @@ export interface WorkspaceRuntimeExecOptions stdin?: Uint8Array | string; timeoutMs?: number; sync?: "wait" | "defer"; + /** + * Strings the app attaches to the run for host modules, such as the + * conversation it belongs to. Host module calls read it as + * `context.execution.metadata`, and `pending()` lists it with each + * waiting call. The run's code never sees it: it isn't part of `env` + * or `input`. A replayable run keeps it when it resumes. Backends that + * run no host modules ignore it. + */ + metadata?: Record; /** * What happens to this run if the Durable Object restarts while it * runs, as on a deploy or crash. Overrides the backend's @@ -308,6 +330,7 @@ export interface ModuleExecutionInput { timeoutMs?: number; sync?: "wait" | "defer"; onInterrupt?: WorkspaceRuntimeInterruptPolicy; + metadata?: Record; } /** @@ -333,6 +356,8 @@ export interface WorkspaceRuntimePendingCall { readonly payload: WorkspaceRuntimeValue; /** When the call started waiting, in milliseconds since the epoch. */ readonly since: number; + /** The `metadata` the run was started with, or an empty object. */ + readonly metadata: WorkspaceExecutionMetadata; } export interface ModuleExecutionEnvelope { diff --git a/packages/computer/src/stub.test.ts b/packages/computer/src/stub.test.ts index 3247d388..adb829bf 100644 --- a/packages/computer/src/stub.test.ts +++ b/packages/computer/src/stub.test.ts @@ -359,7 +359,14 @@ describe("WorkspaceStub", () => { it("passes runtime.pending and runtime.respond through", async () => { await withStub(async (ws) => { - const call = { id: "run", backend: "b", callId: "run:1", payload: { kind: "x" }, since: 1 }; + const call = { + id: "run", + backend: "b", + callId: "run:1", + payload: { kind: "x" }, + since: 1, + metadata: { conversation: "c1" }, + }; vi.spyOn(ws.runtime, "pending").mockResolvedValue([call]); const respond = vi.spyOn(ws.runtime, "respond").mockResolvedValue(); await expect(ws.stub().runtime.pending()).resolves.toEqual([call]); diff --git a/packages/computer/src/tools/ai-sdk/bindings.test.ts b/packages/computer/src/tools/ai-sdk/bindings.test.ts index 7d448258..e03a1574 100644 --- a/packages/computer/src/tools/ai-sdk/bindings.test.ts +++ b/packages/computer/src/tools/ai-sdk/bindings.test.ts @@ -15,7 +15,7 @@ function callContext(): WorkspaceModuleCallContext { deadline: Date.now() + 60_000, access: "read-write", callId: "test-call", - execution: { id: "test-run" }, + execution: { id: "test-run", metadata: {} }, interrupt: () => Promise.reject(new Error("not used")), resolvePath: async (path) => path, }; diff --git a/packages/computer/src/tools/pi-ai/bindings.test.ts b/packages/computer/src/tools/pi-ai/bindings.test.ts index d6736d0e..09d9c650 100644 --- a/packages/computer/src/tools/pi-ai/bindings.test.ts +++ b/packages/computer/src/tools/pi-ai/bindings.test.ts @@ -18,7 +18,7 @@ function callContext( deadline: Date.now() + 60_000, access, callId: "test-call", - execution: { id: "test-run" }, + execution: { id: "test-run", metadata: {} }, interrupt: () => Promise.reject(new Error("not used")), resolvePath: async (path) => path, }; diff --git a/packages/computer/tests/script-runner-worker.ts b/packages/computer/tests/script-runner-worker.ts index 443cdbc3..50bd0325 100644 --- a/packages/computer/tests/script-runner-worker.ts +++ b/packages/computer/tests/script-runner-worker.ts @@ -7,6 +7,7 @@ import { } from "../src/backends/worker-javascript/index"; import { createGitClient } from "../src/git/index"; import type { + WorkspaceExecutionMetadata, WorkspaceRuntimePendingCall, WorkspaceRuntimeValue, WorkspaceStub, @@ -71,7 +72,10 @@ let gateOpen = false; // Named gates a run can wait on after the first, for a second restart. const opened = new Set(); // The execution each executionGate call was told it belongs to. -const executions: Array<{ id: string }> = []; +const executions: Array<{ + id: string; + metadata: WorkspaceExecutionMetadata; +}> = []; function countCall(name: string): number { const count = (durableCalls.get(name) ?? 0) + 1; @@ -136,12 +140,16 @@ function javascriptBackend( if (!opened.has(name)) await new Promise(() => undefined); return "open"; }, + // The execution the call belongs to. + async execution(_args, ctx) { + return { ...ctx.execution, metadata: { ...ctx.execution.metadata } }; + }, // The gate, noting which execution each call belongs to. async executionGate(_args, ctx) { - executions.push({ ...ctx.execution }); + executions.push({ ...ctx.execution, metadata: { ...ctx.execution.metadata } }); countCall("gate"); if (!gateOpen) await new Promise(() => undefined); - return { ...ctx.execution }; + return { ...ctx.execution, metadata: { ...ctx.execution.metadata } }; }, // The gate, for a call that mustn't run twice. unsafeGate: unsafeToRepeat(async () => { @@ -306,6 +314,7 @@ export class HostDO extends DurableObject { id: string; onInterrupt?: "fail" | "restart" | "replay"; network?: boolean; + metadata?: Record; }) { gateOpen = false; opened.clear(); @@ -316,6 +325,7 @@ export class HostDO extends DurableObject { backend: input.network === true ? NETWORK_BACKEND : "worker-javascript", id: input.id, onInterrupt: input.onInterrupt, + metadata: input.metadata, }); void handle.result().catch(() => undefined); // Let the run reach the gate. @@ -330,6 +340,7 @@ export class HostDO extends DurableObject { id: string; onInterrupt?: "fail" | "restart" | "replay"; timeoutMs?: number; + metadata?: Record; }) { durableCalls.clear(); waits.length = 0; @@ -338,6 +349,7 @@ export class HostDO extends DurableObject { id: input.id, onInterrupt: input.onInterrupt ?? "replay", timeoutMs: input.timeoutMs, + metadata: input.metadata, }); void handle.result().catch(() => undefined); return handle.id; @@ -587,6 +599,22 @@ export default class extends WorkerEntrypoint { ); } + // Runs code through the Workspace stub, as another Worker would. + if (url.pathname === "/runtime-stub") { + const body = (await request.json()) as { + source: string; + metadata?: Record; + }; + // SAFETY: The object's getWorkspace returns a WorkspaceStub; RPC typing can't follow it that deep. + using workspace = (await stub.getWorkspace()) as unknown as WorkspaceStub; + using handle = await workspace.runtime.exec(body.source, { + backend: "worker-javascript", + encoding: "utf8", + metadata: body.metadata, + }); + return Response.json(await handle.result()); + } + if (url.pathname === "/runtime-start") { return Response.json({ id: await stub.startRuntime((await request.json()) as { source: string; id: string }), @@ -612,6 +640,7 @@ export default class extends WorkerEntrypoint { id: string; onInterrupt?: "fail" | "restart" | "replay"; network?: boolean; + metadata?: Record; }, ), }); @@ -635,6 +664,7 @@ export default class extends WorkerEntrypoint { id: string; onInterrupt?: "fail" | "restart" | "replay"; timeoutMs?: number; + metadata?: Record; }, ), }); diff --git a/packages/computer/tests/script-runner.test.ts b/packages/computer/tests/script-runner.test.ts index d00bd691..83cc1e75 100644 --- a/packages/computer/tests/script-runner.test.ts +++ b/packages/computer/tests/script-runner.test.ts @@ -334,6 +334,7 @@ describe("WorkspaceRuntime", () => { "approveLater", "delete", "echo", + "execution", "executionGate", "gate", "hold", @@ -349,6 +350,41 @@ describe("WorkspaceRuntime", () => { ]); }); + it("hands host modules a run's metadata and keeps it from the run's code", async () => { + const response = await SELF.fetch("https://example.test/runtime-stub", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + source: ` + import { execution } from "codemode:test-host"; + export default async (input) => ({ + execution: await execution(), + seen: JSON.stringify({ input, env: process.env, argv: process.argv, globals: Object.keys(globalThis) }), + }); + `, + metadata: { conversation: "secret-conversation" }, + }), + }); + const text = await response.text(); + expect(response.status, text).toBe(200); + const { value } = JSON.parse(text) as { + value: { execution: { metadata: unknown }; seen: string }; + }; + expect(value.execution).toMatchObject({ + metadata: { conversation: "secret-conversation" }, + }); + expect(value.seen).not.toContain("secret-conversation"); + }); + + it("gives host modules empty metadata when a run has none", async () => { + const response = await runtime({ + source: `import { execution } from "codemode:test-host"; export default () => execution();`, + }); + const text = await response.text(); + expect(response.status, text).toBe(200); + expect(JSON.parse(text).result.value.metadata).toEqual({}); + }); + it("fails to link an import the host module does not export", async () => { const response = await runtime({ source: ` @@ -996,7 +1032,12 @@ describe("durable executions", () => { async function start( id: string, - body: { source: string; onInterrupt?: string; network?: boolean }, + body: { + source: string; + onInterrupt?: string; + network?: boolean; + metadata?: Record; + }, ) { const response = await SELF.fetch("https://example.test/durable-start", { method: "POST", @@ -1096,8 +1137,12 @@ describe("durable executions", () => { export default async () => executionGate(); `, onInterrupt: "replay", + metadata: { conversation: "c1" }, }); - const execution = { id: "durable-execution" }; + const execution = { + id: "durable-execution", + metadata: { conversation: "c1" }, + }; expect(await result("durable-execution")).toMatchObject({ status: "completed", value: execution, @@ -1365,7 +1410,14 @@ describe("durable executions", () => { }); describe("interrupts", () => { - type Pending = { id: string; backend: string; callId: string; payload: unknown; since: number }; + type Pending = { + id: string; + backend: string; + callId: string; + payload: unknown; + since: number; + metadata: Record; + }; async function start(id: string, source: string, options: Record = {}) { const response = await SELF.fetch("https://example.test/interrupt-start", { @@ -1436,6 +1488,7 @@ describe("interrupts", () => { backend: "worker-javascript", callId: expect.stringMatching(/^wait-answer:\d+$/), payload: { kind: "approval", action: "deploy" }, + metadata: {}, }); await expect(respond("wait-answer", call.callId, { ok: true })).resolves.toEqual({ ok: true }); expect(await result("wait-answer")).toMatchObject({ @@ -1453,11 +1506,12 @@ describe("interrupts", () => { }); it("keeps waiting across a restart", async () => { - await start("wait-restart", approval); - await waitingCall("wait-restart"); + await start("wait-restart", approval, { metadata: { conversation: "c2" } }); + expect((await waitingCall("wait-restart")).metadata).toEqual({ conversation: "c2" }); await SELF.fetch("https://example.test/durable-restart", { method: "POST" }); const call = await waitingCall("wait-restart"); expect(call.payload).toEqual({ kind: "approval", action: "deploy" }); + expect(call.metadata).toEqual({ conversation: "c2" }); await respond("wait-restart", call.callId, { ok: false }); expect(await result("wait-restart")).toMatchObject({ status: "completed",