diff --git a/.changeset/codemode-detach.md b/.changeset/codemode-detach.md new file mode 100644 index 00000000..77285526 --- /dev/null +++ b/.changeset/codemode-detach.md @@ -0,0 +1,5 @@ +--- +"@cloudflare/computer": patch +--- + +`CodemodeBackend` takes an experimental `onDetachedExit`, and `workspace.runtime.detachExec(id)` leaves an exec running with nothing waiting on it, so its exit goes to `onDetachedExit`, saved with the exit and delivered again after a restart or from `runtime.recover()`; see [Detaching](https://github.com/cloudflare/computer/blob/main/docs/05_runtime_interface.md#detaching). diff --git a/.changeset/exec-detach.md b/.changeset/exec-detach.md new file mode 100644 index 00000000..708d7b0c --- /dev/null +++ b/.changeset/exec-detach.md @@ -0,0 +1,5 @@ +--- +"@cloudflare/computer": patch +--- + +The `exec` tool takes an experimental `detach: true`, or `{ afterMs }`, that returns once a run waits for an answer, or has gone on that long, with its `executionId` and output so far, and leaves it running for its backend's `onDetachedExit`; see [Detaching](https://github.com/cloudflare/computer/blob/main/docs/09_tool_interface.md#detaching). diff --git a/docs/05_runtime_interface.md b/docs/05_runtime_interface.md index 08fc0eb3..07710827 100644 --- a/docs/05_runtime_interface.md +++ b/docs/05_runtime_interface.md @@ -167,4 +167,35 @@ Module backends use host capability calls against the authoritative Workspace an `worker-shell` intentionally preserves one-call, buffered-result behavior in this release. It does not retain executions for later reattachment or disposal. `timeoutMs` and a concurrent `killExec()` for a caller-supplied execution ID cooperatively abort just-bash at statement boundaries; by the time an ordinary `exec()` promise returns, the command has already settled. Use the Container or JavaScript isolate when detached execution and retained lifecycle are required. +## Detaching + +Whoever starts an exec usually waits on it for its result. Sometimes that's the wrong thing to do: an exec waiting for a person in [`ctx.interrupt()`](./17_isolate_javascript.md#waiting-for-an-answer) can take hours, and an agent waiting on it can't tell anyone what's happening. Detaching an exec says that nothing waits on it any more. The exec keeps going, and when it exits, its backend hands how it ended to `onDetachedExit`. Only `CodemodeBackend` can detach so far, and only when it has `onDetachedExit`. Detaching is experimental. + +```ts +const workspace = new Workspace({ + storage: ctx.storage, + backends: [ + new CodemodeBackend({ + loader: env.LOADER, + onInterrupt: "replay", + onDetachedExit: async (exec) => { + // exec: { id, backend, status, exitCode, result?, metadata, stdout, stderr, exitedAt } + await conversations.get(exec.metadata.conversation).execEnded(exec); + }, + }), + ], +}); + +const exec = await workspace.runtime.exec(source, { metadata: { conversation: "c1" } }); +if (!(await workspace.runtime.detachExec(exec.id))) { + // It had already exited: read its result as usual. +} +``` + +`detachExec()` returns `false`, and changes nothing, if the exec has already exited. It checks and marks the exec in one step, so every exec ends up either attached, with its result going to whoever reads it, or detached, with its exit going to `onDetachedExit`, never both and never neither. The [`exec` tool](./09_tool_interface.md#detaching) detaches this way when it's built with `detach`. `detachExec()` throws for an exec on a backend that can't detach, and `runtime.backends()` lists which backends can, as `detachable`. + +`status` is `"completed"`, `"failed"` or `"cancelled"`, and `stdout` and `stderr` hold the last 16 KiB of each. The backend saves this in the same transaction that ends the exec, and deletes it once `onDetachedExit` returns. If `onDetachedExit` throws, or the object restarts before it returns, it's called again the next time the backend connects or `runtime.recover()` runs. So `onDetachedExit` may hear about an exec more than once, and should treat its `id` as the key. A quick `onDetachedExit` that saves the exit to the app's own storage, and acts on it from there, keeps the window for a repeat small. Execs that were never detached save nothing. + +Nothing wakes a Durable Object after a deploy, so an app with detached execs should call `recover()` from an alarm while it has some it hasn't heard back from. + See [16. Execution runtime architecture](./16_code_execution.md) and [17. Isolate JavaScript](./17_isolate_javascript.md). diff --git a/docs/09_tool_interface.md b/docs/09_tool_interface.md index 4b1374f9..1f486a2b 100644 --- a/docs/09_tool_interface.md +++ b/docs/09_tool_interface.md @@ -400,6 +400,24 @@ const tools = createAITools({ A pi caller runs each call itself, so it can instead pass the metadata with the call: `execute(call, { metadata })`. Metadata passed with a call wins over `exec: { metadata }`. A caller of `defineExec` passes `metadata` or `toolCallId` on each `execute` the same way. +### Detaching + +By default, `exec` returns when its run ends. A run that waits for a person in [`ctx.interrupt()`](./17_isolate_javascript.md#waiting-for-an-answer), such as for an approval, can take hours, and the agent's turn waits with it: the agent can't tell anyone what it's doing until the answer comes. `detach` lets the tool return early and leave the run going on its own. It's experimental, and needs a backend that can [detach](./05_runtime_interface.md#detaching), such as a `CodemodeBackend` with `onDetachedExit`, which is where the run's exit goes. + +```ts +const tools = createAITools({ + workspace, + exec: { + metadata: ({ toolCallId }) => ({ conversation: this.name, toolCall: toolCallId }), + detach: true, // or { afterMs: 60_000 } to also detach a run still going after a minute + }, +}); +``` + +With `detach`, the tool returns once a call in the run waits for an answer, and, with `afterMs`, once the run has gone on that long. Either way it detaches the run through `runtime.detachExec()` and replies `{ executionId, status, stdout, stderr, note }`, with a `status` of `"waiting"` or `"running"` and, when waiting, the `waiting` calls' `callId`s and payloads. The tool description and the reply's `note` tell the model the run carries on and its outcome arrives later. Aborting the model turn kills a run that hasn't detached, but leaves one that has alone. A run that ends before it detaches returns as usual, and its exit doesn't go to `onDetachedExit`. Runs on a backend that can't detach, such as shell and container backends, stay attached. A tool with no backend that can detach refuses `detach` when it's built. + +The tool doesn't deliver the outcome itself. The backend's `onDetachedExit` hears it, and the app finds the conversation through the run's `metadata` and gives the agent a new message with it. `detach` needs a runtime with `pending()` and `detachExec()`, which a `Workspace` and `workspace.stub()` have. + Wire this tool carefully: it executes arbitrary shell commands inside the configured backend. Treat its output as untrusted text when including it in later model input. Pass `exec: { backends: {} }` or `readonly: true` when command execution is not part of the agent's job, and list backends explicitly when the Workspace has one the model should not use directly. ## `publish` diff --git a/docs/17_isolate_javascript.md b/docs/17_isolate_javascript.md index 6a3d2a53..4205fcdb 100644 --- a/docs/17_isolate_javascript.md +++ b/docs/17_isolate_javascript.md @@ -229,6 +229,10 @@ Once none of its calls are waiting, the run replays to the call and the function - `respond()` throws `ENOTWAITING` for a call that isn't waiting, such as one already answered, so answering twice changes nothing. - Every answer replays the run from the top. +### Detaching a run that waits + +Whoever started a run can await its result, but a run that waits for a person can outlive them: an agent's tool call may want to move on, or a deploy may end the request that started it. Give `CodemodeBackend` an `onDetachedExit` and its runs can detach, so `onDetachedExit` hears how they end instead, even across deploys: see [Detaching](./05_runtime_interface.md#detaching). The `exec` tool detaches a run once it waits, when it's built with `detach`: see [Detaching](./09_tool_interface.md#detaching). + ### Limits - A run gets three attempts, counted again each time it journals a call, and has to finish within its original timeout, counted from when it first started. diff --git a/examples/rlm/worker/executor-tool.ts b/examples/rlm/worker/executor-tool.ts index ec5e53dd..0123c390 100644 --- a/examples/rlm/worker/executor-tool.ts +++ b/examples/rlm/worker/executor-tool.ts @@ -42,7 +42,9 @@ export function createExecutorTool( export function sanitizeExecutorOutput(output: ExecToolOutput): ExecToolOutput { const base = { command: output.command, cwd: output.cwd, backend: output.backend }; - if ("error" in output) return { ...base, error: "Generated execution failed." }; + // The executor never detaches, so a reply without an exit code is a + // failure as far as the model is concerned. + if (!("exitCode" in output)) return { ...base, error: "Generated execution failed." }; return { ...base, diff --git a/packages/computer/README.md b/packages/computer/README.md index aa24839d..f9f70adb 100644 --- a/packages/computer/README.md +++ b/packages/computer/README.md @@ -32,6 +32,8 @@ removed in a future release; each one names its replacement. | `ContainerBackend` and `withWorkspaceContainer` | `@cloudflare/computer/backends/container` | Experimental | | `LegacyContainerBackend` and `withLegacyWorkspaceContainer` | `@cloudflare/computer/backends/container-legacy` | Deprecated: use `ContainerBackend` | | Resuming interrupted isolate JavaScript: `onInterrupt: "restart"` and `"replay"`, `workspace.runtime.recover()`, `unsafeToRepeat()`, and waiting for answers with `ctx.interrupt()`, `pending()` and `respond()` | `@cloudflare/computer`, `@cloudflare/computer/backends/codemode` | Experimental | +| Detaching execs: `CodemodeBackend`'s `onDetachedExit` and `workspace.runtime.detachExec()` | `@cloudflare/computer`, `@cloudflare/computer/backends/codemode` | Experimental | +| Detaching the `exec` tool from a run that waits, with `exec: { detach }` | `@cloudflare/computer/tools/*` | Experimental | | `WorkerShellBackend` and the shell command groups | `@cloudflare/computer/backends/worker-shell`, `@cloudflare/computer/shell/*` | Experimental | | The `codemode:container` host module | `@cloudflare/computer/modules/container` | Experimental | | The `codemode:browser` host module | `@cloudflare/computer/modules/browser` | Experimental | diff --git a/packages/computer/src/backends/codemode/codemode.test.ts b/packages/computer/src/backends/codemode/codemode.test.ts index ba634f63..052b16fd 100644 --- a/packages/computer/src/backends/codemode/codemode.test.ts +++ b/packages/computer/src/backends/codemode/codemode.test.ts @@ -4,7 +4,11 @@ import { describe, expect, it, vi } from "vitest"; import { createGitModule } from "../../modules/git"; import { createCodemodeModule, type WorkspaceModuleScopeEnd } from "../../runtime/codemode-module"; -import type { WorkspaceModule, WorkspaceRuntimeLoader } from "../../runtime/types"; +import type { + WorkspaceModule, + WorkspaceRuntimeExecExit, + WorkspaceRuntimeLoader, +} from "../../runtime/types"; import { Workspace } from "../../workspace"; import { runtimeCode, CodemodeBackend } from "./codemode"; import { ExecutionJournal, fingerprint, loadResume, type ResumeStart, saveResume } from "./journal"; @@ -233,6 +237,108 @@ describe("CodemodeBackend", () => { expect(workerDisposals).toBe(1); }); + describe("detaching", () => { + // A loader whose runs return `value` once `release` is called, so a + // test can act while the run is still going. + function gatedLoader(value: unknown) { + let release!: () => void; + const gate = new Promise((resolve) => (release = resolve)); + const load = vi.fn(() => ({ + getEntrypoint() { + return { + evaluate: async ( + _input: unknown, + host: { + assertResult(value: unknown): Promise; + attachOutput(readable: ReadableStream): Promise; + }, + ) => { + await gate; + await evaluateResult(host, value); + }, + }; + }, + })); + return { load, release: () => release() }; + } + + async function detachedWorkspace(value: unknown, detachable = true) { + const storage = new SQLiteTestStorage(); + const exits: WorkspaceRuntimeExecExit[] = []; + const { load, release } = gatedLoader(value); + const workspace = new Workspace({ + storage, + backends: [ + new CodemodeBackend({ + loader: { load }, + ...(detachable ? { onDetachedExit: (exec) => void exits.push(exec) } : {}), + }), + ], + }); + await workspace.fs.mkdir("/workspace", { recursive: true }); + const savedExits = () => + new Database(storage).scalar( + "SELECT COUNT(*) FROM workspace_runtime_detached_exits", + ); + return { workspace, exits, release, savedExits }; + } + + it("hands onDetachedExit a detached exec's exit, and keeps nothing once it returns", async () => { + const { workspace, exits, release, savedExits } = await detachedWorkspace({ answer: 42 }); + + const handle = await workspace.runtime.exec("export default null", { + metadata: { who: "a" }, + }); + expect(await workspace.runtime.detachExec(handle.id)).toBe(true); + release(); + await handle.result(); + await vi.waitFor(() => expect(exits).toHaveLength(1)); + + expect(exits[0]).toMatchObject({ + id: handle.id, + backend: "codemode", + status: "completed", + exitCode: 0, + result: { answer: 42 }, + metadata: { who: "a" }, + }); + await vi.waitFor(() => expect(savedExits()).toBe(0)); + }); + + it("keeps no exit for an exec nothing detached", async () => { + const { workspace, exits, release, savedExits } = await detachedWorkspace(null); + + const handle = await workspace.runtime.exec("export default null"); + release(); + await handle.result(); + + expect(exits).toEqual([]); + expect(savedExits()).toBe(0); + }); + + it("won't detach an exec that has already exited", async () => { + const { workspace, exits, release } = await detachedWorkspace(null); + + const handle = await workspace.runtime.exec("export default null"); + release(); + await handle.result(); + + expect(await workspace.runtime.detachExec(handle.id)).toBe(false); + expect(exits).toEqual([]); + }); + + it("refuses to detach without onDetachedExit", async () => { + const { workspace, release } = await detachedWorkspace(null, false); + + const handle = await workspace.runtime.exec("export default null"); + + await expect(workspace.runtime.detachExec(handle.id)).rejects.toThrow(/onDetachedExit/); + expect(workspace.runtime.backends()[0]?.detachable).toBe(false); + release(); + await handle.result(); + }); + }); + it("migrates the legacy execution journal schema", async () => { const db = new Database(new SQLiteTestStorage()); initializeSchema(db, () => 0); diff --git a/packages/computer/src/backends/codemode/codemode.ts b/packages/computer/src/backends/codemode/codemode.ts index 2ce2c173..1c8d1951 100644 --- a/packages/computer/src/backends/codemode/codemode.ts +++ b/packages/computer/src/backends/codemode/codemode.ts @@ -29,6 +29,7 @@ import type { WorkspaceRuntimeInterruptPolicy, WorkspaceRuntimeLoader, WorkspaceRuntimePendingCall, + WorkspaceRuntimeExecExit, WorkspaceRuntimeValue, } from "../../runtime/types"; import { WORKSPACE_ROOT } from "../../workspace-root"; @@ -160,6 +161,21 @@ export interface CodemodeBackendOptions { * change in any release. */ onWait?: (call: WorkspaceRuntimePendingCall) => void | Promise; + /** + * Called once a detached exec exits, however it exits, with how it + * ended. An exec detaches through `runtime.detachExec()`, or the exec + * tool's `detach`, when nothing should wait on it any more, such as + * one an agent left waiting for an approval. Nothing else hears how + * it ended, so this is where the outcome goes back to whoever started + * it. Without this, the backend's execs can't detach. + * + * The backend saves the exit in the same transaction that records + * it, and deletes it once this returns. If this throws, or the object + * restarts first, it's called again when the backend next connects or + * from `runtime.recover()`, so it may hear about an exec more than + * once: key on its `id`. Experimental: it may change in any release. + */ + onDetachedExit?: (exec: WorkspaceRuntimeExecExit) => void | Promise; /** * How often to schedule a timer while any execution is running, in * milliseconds. Set it to 0 to turn the timer off. Defaults to 60 @@ -337,12 +353,15 @@ export class CodemodeBackend implements WorkspaceModuleBackend { readonly protocol = "module" as const; readonly type = "codemode"; readonly callable = true; + // Execs detach only when their exits have somewhere to go. + readonly detachable: boolean; readonly id: string; readonly #options: ResolvedCodemodeBackendOptions; readonly #header: string; constructor(options: CodemodeBackendOptions) { this.id = options.id ?? "codemode"; + this.detachable = options.onDetachedExit !== undefined; if (options.egress !== undefined && options.globalOutbound !== undefined) { throw new Error("CodemodeBackend cannot use globalOutbound together with egress."); } @@ -488,6 +507,9 @@ class CodemodeBackendHandle implements WorkspaceModuleBackendHandle { readonly #running = new Map(); readonly #recoveryQueue: Array<() => void> = []; readonly #recovering = new Set(); + // Detached execs whose exit is being handed to `onDetachedExit`, so + // a second delivery pass leaves them alone. + readonly #delivering = new Set(); #keepAlive: ReturnType | undefined; // Zero when the timer is off or can't keep this object in memory. readonly #keepAliveIntervalMs: number; @@ -558,6 +580,17 @@ class CodemodeBackendHandle implements WorkspaceModuleBackendHandle { "INTEGER NOT NULL DEFAULT 0", ); addColumnIfMissing(host, "workspace_runtime_executions", "finished_at", "INTEGER"); + // The run's metadata, as JSON, so its exit can report it for a run + // that saved no start to resume from. + addColumnIfMissing(host, "workspace_runtime_executions", "metadata", "TEXT"); + // Set once nothing waits on the run, so its exit goes to the + // backend's `onDetachedExit`. + addColumnIfMissing( + host, + "workspace_runtime_executions", + "detached", + "INTEGER NOT NULL DEFAULT 0", + ); host.db.run( `UPDATE workspace_runtime_executions SET finished_at = ? @@ -575,6 +608,15 @@ class CodemodeBackendHandle implements WorkspaceModuleBackendHandle { ) `); createResumeTables(host.db); + // Detached execs that exited, until `onDetachedExit` takes them. + host.db.run(` + CREATE TABLE IF NOT EXISTS workspace_runtime_detached_exits ( + backend TEXT NOT NULL, + execution_id TEXT NOT NULL, + data TEXT NOT NULL, + PRIMARY KEY (backend, execution_id) + ) + `); this.#prune(); const interrupted = host.db.all<{ id: string }>( `SELECT id FROM workspace_runtime_executions WHERE backend = ? AND status = 'running'`, @@ -582,11 +624,116 @@ class CodemodeBackendHandle implements WorkspaceModuleBackendHandle { ); for (const { id } of interrupted) this.#resumeInterrupted(id); this.#drainRecoveries(); + void this.#deliverExits(); } - // Recovery runs when the backend connects, so there's nothing left to - // do by the time the runtime asks. - async recover(): Promise {} + // Interrupted runs resume when the backend connects, so all that's + // left is handing on detached execs' exits that haven't been taken. + async recover(): Promise { + await this.#deliverExits(); + } + + // Mark the run detached, unless it has already ended. One transaction, + // so the run can't end between the check and the mark: either its + // exit goes to `onDetachedExit`, or the caller reads its result. + async detachExec(input: { id: string }): Promise { + assertExecutionId(input.id); + if (this.#options.onDetachedExit === undefined) { + throw new Error(`Backend ${this.#backendId} can't detach execs without onDetachedExit.`); + } + const record = this.#recordFor(input.id); + if (record === undefined) { + throw runtimeError("ENOENT", `no such execution: ${input.id}`); + } + if (record.status !== "running") return false; + return this.#host.db.transactionSync(() => { + const row = this.#host.db.one<{ status: string }>( + `SELECT status FROM workspace_runtime_executions WHERE backend = ? AND id = ?`, + this.#backendId, + input.id, + ); + if (row?.status !== "running") return false; + this.#host.db.run( + `UPDATE workspace_runtime_executions SET detached = 1 WHERE backend = ? AND id = ?`, + this.#backendId, + input.id, + ); + return true; + }); + } + + #isDetached(id: string): boolean { + try { + return ( + this.#host.db.one<{ detached: number }>( + `SELECT detached FROM workspace_runtime_executions WHERE backend = ? AND id = ?`, + this.#backendId, + id, + )?.detached === 1 + ); + } catch { + return false; + } + } + + // Hand `onDetachedExit` every saved exit it hasn't taken yet, and + // delete each one it takes. One it throws on stays for the next pass. + async #deliverExits(): Promise { + const { onDetachedExit } = this.#options; + if (onDetachedExit === undefined) return; + let rows: Array<{ execution_id: string; data: string }>; + try { + rows = this.#host.db.all<{ execution_id: string; data: string }>( + `SELECT execution_id, data FROM workspace_runtime_detached_exits WHERE backend = ?`, + this.#backendId, + ); + } catch { + return; + } + await Promise.all( + rows + .filter((row) => !this.#delivering.has(row.execution_id)) + .map(async ({ execution_id: id, data }) => { + this.#delivering.add(id); + try { + // After the current call, so a slow hook can't hold up the + // run that just ended. + await Promise.resolve(); + await onDetachedExit(JSON.parse(data) as WorkspaceRuntimeExecExit); + this.#host.db.run( + `DELETE FROM workspace_runtime_detached_exits WHERE backend = ? AND execution_id = ?`, + this.#backendId, + id, + ); + } catch (error) { + console.error({ message: "onDetachedExit failed", id, error: String(error) }); + } finally { + this.#delivering.delete(id); + } + }), + ); + } + + // What `onDetachedExit` hears about a run that ended with these events. + #execExit( + record: ExecutionRecord, + status: Exclude, + events: readonly WorkspaceRuntimeEvent[], + ): WorkspaceRuntimeExecExit { + const all = [...record.events, ...events]; + const exit = all.filter((event) => event.name === "exit").at(-1); + return { + id: record.id, + backend: this.#backendId, + status, + exitCode: exit?.name === "exit" ? exit.code : 1, + ...(exit?.name === "exit" && exit.result !== undefined ? { result: exit.result } : {}), + metadata: this.#metadataOf(record.id), + stdout: outputTail(all, "stdout"), + stderr: outputTail(all, "stderr"), + exitedAt: Date.now(), + }; + } // Start a run the restart interrupted again, if it asked for that and // still can, or settle it as failed. @@ -836,18 +983,27 @@ class CodemodeBackendHandle implements WorkspaceModuleBackendHandle { })); } - // 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. + // A run's metadata. A running record in memory has it with its + // execution. Otherwise it's read from the run's row, or, for a run + // saved before the row kept it, from its saved start. #metadataOf(id: string): WorkspaceExecutionMetadata { const known = this.#records.get(id)?.execution; if (known !== undefined) return known.metadata; - let saved: ReturnType; + let metadata: Record | undefined; try { - saved = loadResume(this.#host.db, this.#backendId, id); + const row = this.#host.db.one<{ metadata: string | null }>( + `SELECT metadata FROM workspace_runtime_executions WHERE backend = ? AND id = ?`, + this.#backendId, + id, + ); + metadata = + typeof row?.metadata === "string" + ? (JSON.parse(row.metadata) as Record) + : loadResume(this.#host.db, this.#backendId, id)?.start.metadata; } catch { - saved = undefined; + metadata = undefined; } - return this.#execution(id, saved?.start.metadata).metadata; + return this.#execution(id, metadata).metadata; } // What host module calls of a run are told about it, frozen so one @@ -1072,11 +1228,12 @@ class CodemodeBackendHandle implements WorkspaceModuleBackendHandle { try { this.#host.db.run( `INSERT INTO workspace_runtime_executions - (backend, id, status, created_at, finished_at) - VALUES (?, ?, 'running', ?, NULL)`, + (backend, id, status, created_at, finished_at, metadata) + VALUES (?, ?, 'running', ?, NULL, ?)`, this.#backendId, id, Date.now(), + input.metadata === undefined ? null : JSON.stringify(input.metadata), ); } catch (error) { const durable = this.#recordFor(id); @@ -1606,8 +1763,25 @@ class CodemodeBackendHandle implements WorkspaceModuleBackendHandle { record.persistenceFailed = false; let settledStatus = status; let settledEvents = events; + // Only a detached run's exit is saved for `onDetachedExit`: whoever + // waits on any other run hears its result from the run itself. Read + // before the transaction clears the run's saved start, which is + // where a resumed run's metadata may come from. + const exit = + this.#options.onDetachedExit !== undefined && this.#isDetached(record.id) + ? this.#execExit(record, status, events) + : undefined; try { this.#host.db.transactionSync(() => { + if (exit !== undefined) { + this.#host.db.run( + `INSERT OR REPLACE INTO workspace_runtime_detached_exits (backend, execution_id, data) + VALUES (?, ?, ?)`, + this.#backendId, + record.id, + JSON.stringify(exit), + ); + } this.#host.db.run( `UPDATE workspace_runtime_executions SET status = ?, finished_at = ? @@ -1649,6 +1823,7 @@ class CodemodeBackendHandle implements WorkspaceModuleBackendHandle { } catch { // Retention cleanup is retried on the next operation. } + if (exit !== undefined) void this.#deliverExits(); } } @@ -2357,6 +2532,16 @@ function recordedOutput( return bytes; } +const SETTLED_OUTPUT_BYTES = 16 * 1024; + +// The end of one of a run's streams, as text, for a detached run's exit. +function outputTail(events: readonly WorkspaceRuntimeEvent[], name: "stdout" | "stderr"): string { + const bytes = recordedOutput(events, name); + return new TextDecoder().decode( + bytes.subarray(Math.max(0, bytes.byteLength - SETTLED_OUTPUT_BYTES)), + ); +} + function bytesEqual(a: Uint8Array, b: Uint8Array): boolean { if (a.byteLength !== b.byteLength) return false; for (let i = 0; i < a.byteLength; i++) if (a[i] !== b[i]) return false; diff --git a/packages/computer/src/index.ts b/packages/computer/src/index.ts index f3f37b63..b696dfdd 100644 --- a/packages/computer/src/index.ts +++ b/packages/computer/src/index.ts @@ -105,6 +105,7 @@ export type { WorkspaceRuntimeAccess, WorkspaceRuntimeDisposeOptions, WorkspaceRuntimeEvent, + WorkspaceRuntimeExecExit, WorkspaceRuntimeExecHandle, WorkspaceRuntimeExecOptions, WorkspaceRuntimeGetOptions, diff --git a/packages/computer/src/runtime/runtime.ts b/packages/computer/src/runtime/runtime.ts index fde5136c..26333a9c 100644 --- a/packages/computer/src/runtime/runtime.ts +++ b/packages/computer/src/runtime/runtime.ts @@ -53,6 +53,11 @@ export interface WorkspaceBackendInfo { readonly callable: boolean; /** What the backend tells a model about itself. */ readonly description?: string; + /** + * Whether its execs can detach, through `runtime.detachExec()`, such + * as a `CodemodeBackend` with `onDetachedExit`. + */ + readonly detachable: boolean; } export class WorkspaceRuntime { @@ -80,9 +85,14 @@ export class WorkspaceRuntime { protocol: isModuleBackend(backend) ? ("module" as const) : ("command" as const), callable: backend.callable === true, ...(backend.description === undefined ? {} : { description: backend.description }), + detachable: this.#detachable(backend), })); } + #detachable(backend: WorkspaceRegisteredBackend): boolean { + return isModuleBackend(backend) && backend.detachable === true; + } + exec(source: string): Promise>; exec( source: string, @@ -162,10 +172,12 @@ export class WorkspaceRuntime { } /** - * Resume or settle the runs a Durable Object restart interrupted. + * Resume or settle the runs a Durable Object restart interrupted, and + * call a backend's `onDetachedExit` again for exits it hasn't taken + * yet. * Backends do this when they first connect, which the next runtime * call or `ready({ all: true })` triggers. Call this from an alarm to - * resume runs without waiting for one. + * do it without waiting for one. */ async recover(): Promise { for (const [id, backend] of this.#options.backends) { @@ -212,6 +224,31 @@ export class WorkspaceRuntime { await handle.respond({ id, callId, answer }); } + /** + * Note that nothing waits on the exec any more, and leave it running. + * Once it exits, its backend's `onDetachedExit` hears how it ended. + * Returns false, and changes nothing, if the exec has already exited: + * read its result from `getExec()` instead. Needs a backend that can + * detach, such as a `CodemodeBackend` with `onDetachedExit`. + * Experimental: it may change in any release. + */ + async detachExec(id: string, options: { backend?: string } = {}): Promise { + assertExecutionId(id); + const backend = this.#backend(options.backend); + const registered = this.#options.backends.get(backend); + const handle = await this.#options.backendHandle(backend); + if ( + registered === undefined || + !this.#detachable(registered) || + handle.detachExec === undefined + ) { + throw new Error( + `Backend ${backend} can't detach execs: it needs somewhere to send their exits, such as CodemodeBackend's onDetachedExit.`, + ); + } + return handle.detachExec({ id }); + } + async killExec(id: string, options: WorkspaceRuntimeKillOptions = {}): Promise { assertExecutionId(id); const backend = this.#backend(options.backend); diff --git a/packages/computer/src/runtime/types.ts b/packages/computer/src/runtime/types.ts index 2c52a4f5..499b3e3f 100644 --- a/packages/computer/src/runtime/types.ts +++ b/packages/computer/src/runtime/types.ts @@ -369,6 +369,28 @@ export interface WorkspaceRuntimePendingCall { readonly metadata: WorkspaceExecutionMetadata; } +/** + * How a detached exec ended, as a backend's `onDetachedExit` hears it. + * Experimental: it may change in any release. + */ +export interface WorkspaceRuntimeExecExit { + /** The exec that exited. */ + readonly id: string; + readonly backend: string; + readonly status: "completed" | "failed" | "cancelled"; + readonly exitCode: number; + /** What the run's default export returned, when it returned something. */ + readonly result?: WorkspaceRuntimeValue; + /** The `metadata` the run was started with, or an empty object. */ + readonly metadata: WorkspaceExecutionMetadata; + /** The last 16 KiB of the run's standard output. */ + readonly stdout: string; + /** The last 16 KiB of the run's standard error. */ + readonly stderr: string; + /** When the run ended, in milliseconds since the epoch. */ + readonly exitedAt: number; +} + export interface ModuleExecutionEnvelope { id: string; // Identity of the process-local backend runtime that owns this @@ -396,9 +418,13 @@ export interface WorkspaceModuleBackendHandle { }): Promise; killExec(input: { id: string; signal?: KillSignal; runtimeId?: string }): Promise; disposeExec(input: { id: string; runtimeId?: string }): Promise; - // Settle the runs a restart interrupted. A backend that recovers - // when it connects needs nothing more here. + // Settle the runs a restart interrupted, and hand on detached execs' + // exits that `onDetachedExit` hasn't taken yet. A backend that does + // this when it connects only needs to retry the hook here. recover?(): Promise; + // Note that nothing waits on the exec any more, so its exit goes to + // the backend's `onDetachedExit`. False if it has already exited. + detachExec?(input: { id: string }): Promise; // Calls waiting for an answer, and answering one. pending?(): Promise[]>; respond?(input: { id: string; callId: string; answer: WorkspaceRuntimeValue }): Promise; @@ -422,6 +448,11 @@ export interface WorkspaceModuleBackend { readonly callable?: boolean; /** What the backend tells a model about itself. Shown by the exec tool. */ readonly description?: string; + /** + * Whether its execs can detach, through `runtime.detachExec()`: the + * backend supports it and has somewhere to send their exits. + */ + readonly detachable?: boolean; connect(host: WorkspaceModuleBackendHost): Promise; } diff --git a/packages/computer/src/stub.ts b/packages/computer/src/stub.ts index c7951bab..b2616916 100644 --- a/packages/computer/src/stub.ts +++ b/packages/computer/src/stub.ts @@ -473,6 +473,10 @@ export class WorkspaceRuntimeStub extends RpcTarget { return this.#ws.runtime.disposeExec(id, options); } + detachExec(id: string, options: { backend?: string } = {}): Promise { + return this.#ws.runtime.detachExec(id, options); + } + // Plain data both ways, so nothing here needs disposing. pending(): Promise { return this.#ws.runtime.pending(); diff --git a/packages/computer/src/tools/ai-sdk/index.test.ts b/packages/computer/src/tools/ai-sdk/index.test.ts index c86185d8..b4e317fa 100644 --- a/packages/computer/src/tools/ai-sdk/index.test.ts +++ b/packages/computer/src/tools/ai-sdk/index.test.ts @@ -2327,6 +2327,44 @@ describe("createAITools exec metadata", () => { }); }); +describe("createAITools exec detach", () => { + // A codemode run that prints once and doesn't end on its own. + const workspace = { + runtime: { + async exec() { + return { + id: "run-1", + async *[Symbol.asyncIterator]() { + yield { name: "stdout" as const, value: "started\n" }; + await new Promise(() => undefined); + }, + result: () => new Promise(() => undefined), + }; + }, + pending: async () => [], + detachExec: async () => true, + backends: () => [ + { id: "codemode", protocol: "module" as const, callable: true, detachable: true }, + ], + }, + }; + + it("takes detach on the tool set, and says so in the description", async () => { + const tools = createAITools({ + // SAFETY: The test only runs exec, which reads nothing but the runtime. + workspace: workspace as unknown as Parameters[0]["workspace"], + exec: { detach: { afterMs: 20 } }, + }); + + expect(tools.exec?.description).toContain("comes back early"); + expect(await executeTool(tools.exec, { command: "research()" })).toMatchObject({ + executionId: "run-1", + status: "running", + stdout: "started\n", + }); + }); +}); + describe("createAITools publish tool", () => { it("adds publish by default when assets are configured", async () => { const calls: Array<{ path: string; expiresAfter: number; prefix?: string }> = []; diff --git a/packages/computer/src/tools/common/exec.test.ts b/packages/computer/src/tools/common/exec.test.ts index 6ffeefac..153bb161 100644 --- a/packages/computer/src/tools/common/exec.test.ts +++ b/packages/computer/src/tools/common/exec.test.ts @@ -83,3 +83,180 @@ describe("defineExec metadata", () => { expect(metadata).toEqual([]); }); }); + +// A run that prints a line, then stays running until `finish()`, as a +// run waiting for an approval does. `waiting` lists what pending() +// reports for it. detachExec() succeeds while the run is going, after +// `detachDelay` if set, as the runtime's does. +function runningWorkspace( + backends: Array<{ + id: string; + protocol: "module" | "command"; + callable: boolean; + detachable: boolean; + }> = [{ id: "codemode", protocol: "module", callable: true, detachable: true }], +) { + const state = { + killed: 0, + ended: false, + detached: [] as string[], + detachDelay: undefined as Promise | undefined, + waiting: [] as Array<{ id: string; callId: string; payload: unknown }>, + finish: (_code: number) => {}, + }; + const workspace: ExecWorkspaceLike = { + runtime: { + async exec() { + let finish!: (code: number) => void; + const finished = new Promise((resolve) => (finish = resolve)); + state.finish = (code) => { + state.ended = true; + finish(code); + }; + return { + id: "run-1", + async *[Symbol.asyncIterator]() { + yield { name: "stdout" as const, value: "asking\n" }; + yield { name: "exit" as const, code: await finished }; + }, + result: async () => ({ exitCode: await finished, stdout: "asking\n", stderr: "" }), + kill: async () => { + state.killed += 1; + }, + }; + }, + pending: async () => state.waiting, + async detachExec(id) { + const going = !state.ended; + await state.detachDelay; + if (going) state.detached.push(id); + return going; + }, + backends: () => backends, + }, + }; + return { workspace, state }; +} + +describe("defineExec detach", () => { + it("returns once the run waits for an answer, and leaves it running", async () => { + const { workspace, state } = runningWorkspace(); + const exec = defineExec({ workspace, detach: true }); + const output = last(exec.execute({ command: "refund()" })); + state.waiting = [{ id: "run-1", callId: "run-1:2", payload: { kind: "approval" } }]; + + expect(await output).toMatchObject({ + backend: "codemode", + executionId: "run-1", + status: "waiting", + stdout: "asking\n", + waiting: [{ callId: "run-1:2", payload: { kind: "approval" } }], + }); + expect(state.detached).toEqual(["run-1"]); + expect(state.killed).toBe(0); + }); + + it("only counts waits that belong to its own run", async () => { + const { workspace, state } = runningWorkspace(); + state.waiting = [{ id: "someone-else", callId: "someone-else:1", payload: null }]; + const exec = defineExec({ workspace, detach: true }); + const output = last(exec.execute({ command: "refund()" })); + await new Promise((resolve) => setTimeout(resolve, 400)); + state.finish(0); + + expect(await output).toMatchObject({ exitCode: 0, stdout: "asking\n" }); + expect(state.detached).toEqual([]); + }); + + it("returns once the run has gone on for afterMs", async () => { + const { workspace, state } = runningWorkspace(); + const exec = defineExec({ workspace, detach: { afterMs: 50 } }); + + expect(await last(exec.execute({ command: "research()" }))).toMatchObject({ + executionId: "run-1", + status: "running", + stdout: "asking\n", + }); + expect(state.killed).toBe(0); + }); + + it("returns a run that ends before it detaches as usual", async () => { + const { workspace, state } = runningWorkspace(); + const exec = defineExec({ workspace, detach: { afterMs: 5_000 } }); + const output = last(exec.execute({ command: "lookup()" })); + state.finish(0); + + const settled = await output; + expect(settled).toMatchObject({ exitCode: 0, stdout: "asking\n" }); + expect(settled).not.toHaveProperty("status"); + expect(state.detached).toEqual([]); + }); + + it("replies as detached when the run ends while its detach goes through", async () => { + const { workspace, state } = runningWorkspace(); + let release!: () => void; + state.detachDelay = new Promise((resolve) => (release = resolve)); + const exec = defineExec({ workspace, detach: { afterMs: 10 } }); + const output = last(exec.execute({ command: "research()" })); + await new Promise((resolve) => setTimeout(resolve, 30)); + // The runtime took the detach, so the exit goes to the Workspace: + // the tool mustn't hand it to the model as well. + state.finish(0); + release(); + + expect(await output).toMatchObject({ executionId: "run-1", status: "running" }); + expect(state.detached).toEqual(["run-1"]); + }); + + it("stays attached when the run ended before its detach could go through", async () => { + const { workspace, state } = runningWorkspace(); + const exec = defineExec({ workspace, detach: { afterMs: 10 } }); + const output = last(exec.execute({ command: "research()" })); + state.ended = true; + await new Promise((resolve) => setTimeout(resolve, 30)); + state.finish(0); + + const settled = await output; + expect(settled).toMatchObject({ exitCode: 0, stdout: "asking\n" }); + expect(settled).not.toHaveProperty("status"); + }); + + it("doesn't kill a detached run when the turn aborts afterwards", async () => { + const { workspace, state } = runningWorkspace(); + const exec = defineExec({ workspace, detach: { afterMs: 20 } }); + const abort = new AbortController(); + + await last(exec.execute({ command: "research()" }, { abortSignal: abort.signal })); + abort.abort(); + + expect(state.killed).toBe(0); + }); + + it("leaves a run on a backend that can't detach attached", async () => { + const { workspace, state } = runningWorkspace([ + { id: "codemode", protocol: "module", callable: true, detachable: true }, + { id: "sh", protocol: "command", callable: false, detachable: false }, + ]); + const exec = defineExec({ workspace, detach: { afterMs: 20 } }); + const output = last(exec.execute({ command: "sleep 1", backend: "sh" })); + await new Promise((resolve) => setTimeout(resolve, 100)); + state.finish(0); + + const settled = await output; + expect(settled).toMatchObject({ backend: "sh", exitCode: 0 }); + expect(settled).not.toHaveProperty("status"); + expect(exec.description).toContain('A run on "codemode" that waits for an answer'); + }); + + it("refuses to detach when no backend can", () => { + const { workspace } = runningWorkspace([ + { id: "codemode", protocol: "module", callable: true, detachable: false }, + ]); + expect(() => defineExec({ workspace, detach: true })).toThrow(/onDetachedExit/); + }); + + it("refuses to detach without a runtime that can", () => { + const { workspace } = recordingWorkspace(); + expect(() => defineExec({ workspace, detach: true })).toThrow(/detachExec\(\)/); + }); +}); diff --git a/packages/computer/src/tools/common/exec.ts b/packages/computer/src/tools/common/exec.ts index 7b1906b0..2e49e793 100644 --- a/packages/computer/src/tools/common/exec.ts +++ b/packages/computer/src/tools/common/exec.ts @@ -42,6 +42,9 @@ export type ExecStreamEvent = // chunks by iterating the handle when it is async-iterable, and // falls back to draining result() when it is not. export interface ExecRuntimeHandle extends Partial> { + // The execution's id. A detached run reports it, and `pending()` + // lists the run's waiting calls under it. + readonly id?: string; result(): Promise<{ exitCode: number; stdout: string; @@ -69,6 +72,12 @@ export interface ExecWorkspaceLike { output?: { maxBytes: number; maxLines: number }; }, ): Promise; + // Calls waiting in ctx.interrupt(), by execution id, and noting + // that nothing waits on a run any more. Needed to detach. + pending?(): Promise< + ReadonlyArray<{ readonly id: string; readonly callId: string; readonly payload: unknown }> + >; + detachExec?(id: string, options?: { backend?: string }): Promise; // What each registered backend says about itself: whether it takes // structured `input`, and its description for the model. Used to // build the tool, to offer every backend when the caller picks none, @@ -101,6 +110,15 @@ export type ExecMetadata = | Record | ((call: { toolCallId: string }) => Record); +/** + * Whether the exec tool may return before a run ends, leaving it + * running. `true` detaches a run once it waits for an answer in + * `ctx.interrupt()`, such as for an approval. `{ afterMs }` also + * detaches a run that is still going after that many milliseconds. + * Experimental: it may change in any release. + */ +export type ExecDetachOptions = true | { readonly afterMs?: number }; + export interface ExecToolOptions { workspace: ExecWorkspaceLike; // Omit to offer every backend the Workspace has. With one backend @@ -124,6 +142,16 @@ export interface ExecToolOptions { * belongs to. Metadata a call passes itself, as a pi caller can, wins. */ metadata?: ExecMetadata; + /** + * Return before a run ends, and leave it running, so the agent can + * carry on. The reply has the run's `executionId`, a `status` of + * `"waiting"` or `"running"`, and its output so far. Aborting the + * model turn doesn't kill a run once it has detached. The run's exit + * goes to its backend's `onDetachedExit`, which tells the agent. Needs + * a backend that can detach, such as a `CodemodeBackend` with + * `onDetachedExit`. + */ + detach?: ExecDetachOptions; // Clock backing the running-snapshot coalescing floor. Defaults to // Date.now; injectable so tests can drive the interval deterministically. now?: () => number; @@ -148,6 +176,19 @@ export type ExecToolOutput = stderr: string; result?: unknown; } + | { + command: string; + cwd: string | null; + backend: string; + // The run is still going: it detached, as `ExecToolOptions.detach` asked. + executionId: string; + status: "waiting" | "running"; + stdout: string; + stderr: string; + // The calls waiting for an answer, when the run is waiting. + waiting?: Array<{ callId: string; payload: unknown }>; + note: string; + } | { command: string; cwd: string | null; backend: string; error: string }; export interface ExecInput { @@ -194,6 +235,11 @@ export function defineExec(options: ExecToolOptions): ExecDefinition { ); const now = options.now ?? Date.now; const runtime = options.workspace.runtime; + const detach = options.detach; + const afterMs = typeof detach === "object" ? detach.afterMs : undefined; + if (afterMs !== undefined && (!Number.isFinite(afterMs) || afterMs < 0)) { + throw new Error("createExecTool: detach.afterMs must be a finite number of milliseconds"); + } // Read the backends once, so selection and descriptions agree. const known = runtime.backends?.(); const selected = selectBackends(options.backends, known); @@ -208,10 +254,34 @@ export function defineExec(options: ExecToolOptions): ExecDefinition { const text = [guidance, own].filter((part) => part !== undefined && part !== "").join("\n\n") || (callable ? "Runs `command` as module source." : "Runs shell commands."); - return { id, text, callable }; + return { id, text, callable, detachable: info?.detachable === true }; }); const callableBackendIds = new Set(backends.filter((b) => b.callable).map((b) => b.id)); - const description = describeTool(backends, limits); + // Only runs on a backend that can detach do: their exit goes to the + // backend's `onDetachedExit`. Runs elsewhere stay attached. + const detachable = new Set( + detach === undefined ? [] : backends.filter((b) => b.detachable).map((b) => b.id), + ); + const detachExec = runtime.detachExec; + const pending = runtime.pending; + if (detach !== undefined) { + if (detachExec === undefined || pending === undefined) { + throw new Error("createExecTool: detach needs a runtime with detachExec() and pending()"); + } + if (detachable.size === 0) { + throw new Error( + "createExecTool: detach needs a backend that can detach, such as a CodemodeBackend with onDetachedExit", + ); + } + } + const description = [ + describeTool(backends, limits), + detach === undefined + ? "" + : detachHint(afterMs, detachable.size === backends.length ? undefined : [...detachable]), + ] + .filter((part) => part !== "") + .join("\n\n"); // Offer only the fields that can work: `backend` when there is a // choice, `input` when some backend accepts it. const shape: Record = { @@ -245,6 +315,76 @@ export function defineExec(options: ExecToolOptions): ExecDefinition { // SAFETY: Every field in `shape` has the type ExecInput gives it, and the fields left out are optional there. const inputSchema = z.object(shape) as unknown as z.ZodType; + // Settles when the run should detach, as `detach` asks, and never + // otherwise. Stop it once the tool is done with the run. The run is + // marked detached before this settles; if it ended first, this never + // settles and the tool reads its result as usual. + function watchDetach( + handle: ExecRuntimeHandle, + backend: string, + ): { + promise: Promise; + // Before the tool returns a result: whether a detach already under + // way went through. If it did, the run's exit goes to the Workspace, + // so the tool must reply as detached rather than with the result. + settle(): Promise; + stop(): void; + } { + const timers: Array> = []; + let stopped = false; + let inFlight: Promise | undefined; + const promise = new Promise((resolve) => { + const id = handle.id; + if (!detachable.has(backend) || id === undefined) return; + if (detachExec === undefined || pending === undefined) return; + // Detach once, if the run hasn't ended. A run that can't detach + // stays attached rather than losing its result. + const tryDetach = async (reason: DetachReason) => { + if (inFlight !== undefined || stopped) return; + inFlight = detachExec + .call(runtime, id, { backend }) + .catch((error: unknown) => { + console.error({ message: "exec: detach failed", id, error: String(error) }); + return false; + }) + .then((detached) => (detached ? reason : undefined)); + const detached = await inFlight; + if (detached !== undefined) resolve(detached); + else stop(); + }; + if (afterMs !== undefined) { + timers.push(setTimeout(() => void tryDetach({ status: "running" }), afterMs)); + } + const poll = async () => { + if (stopped || inFlight !== undefined) return; + const waiting = await pending.call(runtime).catch(() => []); + const own = waiting.filter((call) => call.id === id); + if (own.length > 0) { + await tryDetach({ + status: "waiting", + waiting: own.map((call) => ({ callId: call.callId, payload: call.payload })), + }); + return; + } + if (!stopped) timers.push(setTimeout(() => void poll(), WAIT_POLL_MS)); + }; + void poll(); + }); + function stop() { + stopped = true; + for (const timer of timers) clearTimeout(timer); + } + return { + promise, + async settle() { + const reason = await inFlight; + stop(); + return reason; + }, + stop, + }; + } + return { description, inputSchema, @@ -290,15 +430,34 @@ export function defineExec(options: ExecToolOptions): ExecDefinition { // Aborting the model turn kills the backend execution so it does // not run on unobserved after the tool stops iterating. The run // then emits its terminal event and the stream closes normally. + // A run that detached is meant to run on, so it's left alone. const onAbort = () => void handle.kill?.().catch(() => undefined); if (abortSignal?.aborted) onAbort(); else abortSignal?.addEventListener("abort", onAbort, { once: true }); + const detaching = watchDetach(handle, selectedBackend); try { yield* runExecution(); } finally { + detaching.stop(); abortSignal?.removeEventListener("abort", onAbort); } + // The snapshot of a run that detached, which carries on without the tool. + function detached(reason: DetachReason, stdout: string, stderr: string): ExecToolOutput { + return { + ...base, + executionId: handle.id ?? "", + status: reason.status, + stdout, + stderr, + ...(reason.status === "waiting" ? { waiting: reason.waiting } : {}), + note: + reason.status === "waiting" + ? "The run is still going. It is waiting for an answer to the calls in `waiting`, and carries on by itself once they are answered. Don't run it again: you'll get a message with its outcome when it ends." + : "The run is still going in the background. Don't run it again: you'll get a message with its outcome when it ends.", + }; + } + // Produce the run's snapshots. Streams the raw events when the // handle is iterable; otherwise drains the aggregate result. async function* runExecution(): AsyncGenerator { @@ -317,8 +476,25 @@ export function defineExec(options: ExecToolOptions): ExecDefinition { // chatty command would otherwise yield a full-buffer snapshot // per chunk; the terminal snapshot below always fires. let lastSnapshot = 0; + const iterator = (handle as AsyncIterable)[Symbol.asyncIterator](); try { - for await (const event of handle as AsyncIterable) { + while (true) { + const next = iterator.next(); + const step = await Promise.race([ + next.then((value) => ({ kind: "event" as const, value })), + detaching.promise.then((reason) => ({ kind: "detach" as const, reason })), + ]); + if (step.kind === "detach") { + // Stop listening without stopping the run. The read in + // flight settles whenever the run next prints or ends. + next.catch(() => undefined); + void iterator.return?.()?.catch(() => undefined); + abortSignal?.removeEventListener("abort", onAbort); + yield detached(step.reason, stdout.render(), stderr.render()); + return; + } + if (step.value.done === true) break; + const event = step.value.value; if (event.name === "stdout") stdout.push(event.value); else if (event.name === "stderr") stderr.push(event.value); else { @@ -344,6 +520,12 @@ export function defineExec(options: ExecToolOptions): ExecDefinition { yield { ...base, error: errorMessage(err) }; return; } + const late = await detaching.settle(); + if (late !== undefined) { + abortSignal?.removeEventListener("abort", onAbort); + yield detached(late, stdout.render(), stderr.render()); + return; + } yield { ...base, exitCode, @@ -356,7 +538,23 @@ export function defineExec(options: ExecToolOptions): ExecDefinition { // Non-streaming handle: drain the aggregate result. try { - const result = await handle.result(); + const settled = handle.result(); + const step = await Promise.race([ + settled.then((value) => ({ kind: "result" as const, value })), + detaching.promise.then((reason) => ({ kind: "detach" as const, reason })), + ]); + if (step.kind === "detach") { + settled.catch(() => undefined); + abortSignal?.removeEventListener("abort", onAbort); + yield detached(step.reason, "", ""); + return; + } + const result = step.value; + const late = await detaching.settle(); + if (late !== undefined) { + yield detached(late, "", ""); + return; + } yield { ...base, exitCode: result.exitCode, @@ -372,6 +570,29 @@ export function defineExec(options: ExecToolOptions): ExecDefinition { }; } +// How often a run that may detach on a wait checks for one. +const WAIT_POLL_MS = 250; + +type DetachReason = + | { status: "running" } + | { status: "waiting"; waiting: Array<{ callId: string; payload: unknown }> }; + +function detachHint( + afterMs: number | undefined, + // The backends whose runs detach, when not every one does. + only: readonly string[] | undefined, +): string { + const when = [ + "waits for an answer, such as an approval", + ...(afterMs === undefined + ? [] + : [`is still going after ${Math.round(afterMs / 1000)} seconds`]), + ]; + const where = + only === undefined ? "" : ` on ${only.map((id) => JSON.stringify(id)).join(" or ")}`; + return `A run${where} that ${when.join(" or ")} comes back early with its \`executionId\`, a \`status\` of "waiting" or "running", and its output so far, and keeps going on its own. Tell the user what is happening and carry on; don't run it again. Its outcome arrives in a later message.`; +} + const FILE_TOOLS_HINT = "Prefer the dedicated read, write, and edit tools for file operations."; // How the reply cuts long output, and where the rest goes. diff --git a/packages/computer/src/tools/common/options.ts b/packages/computer/src/tools/common/options.ts index 27e99e09..f55ac8a5 100644 --- a/packages/computer/src/tools/common/options.ts +++ b/packages/computer/src/tools/common/options.ts @@ -1,4 +1,10 @@ -import type { ExecBackends, ExecMetadata, ExecToolOptions, ExecWorkspaceLike } from "./exec"; +import type { + ExecBackends, + ExecDetachOptions, + ExecMetadata, + ExecToolOptions, + ExecWorkspaceLike, +} from "./exec"; import { confineWorkspace } from "./fs/confine"; import type { EditToolOptions } from "./fs/edit"; import type { ReadToolOptions } from "./fs/read"; @@ -20,8 +26,8 @@ export interface CreateToolsOptions { /** * Options for the `exec` tool. Omit `exec` or its `backends` to offer * every backend the Workspace has; `{ backends: {} }` means no exec - * tool. `exec: {}` and keys other than `backends`, `output`, and - * `metadata`, such as a backend map from before `exec.backends`, + * tool. `exec: {}` and keys other than `backends`, `output`, + * `metadata` and `detach`, such as a backend map from before `exec.backends`, * throw. */ exec?: ExecToolSetOptions; @@ -50,6 +56,12 @@ export interface ExecToolSetOptions { * sees it. */ metadata?: ExecMetadata; + /** + * Return before a run ends and leave it running, once it waits for an + * answer or, with `afterMs`, has gone on that long. Its exit goes to + * the backend's `onDetachedExit`. Experimental. + */ + detach?: ExecDetachOptions; } interface LegacyShellOptions extends Omit { @@ -90,7 +102,7 @@ export function resolveToolOptions(options: CreateToolsOptions): ResolvedToolOpt }; } -const EXEC_KEYS = new Set(["backends", "output", "metadata"]); +const EXEC_KEYS = new Set(["backends", "output", "metadata", "detach"]); // Reject the shapes `exec` had before it took `{ backends, output, // metadata }`, rather than offering every backend for them. @@ -99,7 +111,7 @@ function checkExec(exec: ExecToolSetOptions | undefined): void { const keys = Object.keys(exec); if (keys.some((key) => !EXEC_KEYS.has(key))) { throw new TypeError( - "exec takes `{ backends, output, metadata }`; move backend options under `exec.backends`", + "exec takes `{ backends, output, metadata, detach }`; move backend options under `exec.backends`", ); } if (keys.length === 0) { @@ -115,13 +127,14 @@ function execOptions(options: CreateToolsOptions): ExecToolOptions | undefined { if (runtime === undefined) return undefined; const selected = selectExec(options, runtime); if (Object.keys(selected.backends).length === 0) return undefined; - const { output, metadata } = options.exec ?? {}; + const { output, metadata, detach } = options.exec ?? {}; return { workspace: { runtime }, ...selected, ...(output?.maxBytes === undefined ? {} : { maxBytes: output.maxBytes }), ...(output?.maxLines === undefined ? {} : { maxLines: output.maxLines }), ...(metadata === undefined ? {} : { metadata }), + ...(detach === undefined ? {} : { detach }), }; } diff --git a/packages/computer/src/tools/tanstack-ai/index.test.ts b/packages/computer/src/tools/tanstack-ai/index.test.ts index 1458b385..c5b84100 100644 --- a/packages/computer/src/tools/tanstack-ai/index.test.ts +++ b/packages/computer/src/tools/tanstack-ai/index.test.ts @@ -360,6 +360,45 @@ describe("createTanStackTools", () => { await workspace.close(); }); + it("returns a detached reply rather than emitting it as a running snapshot", async () => { + const workspace = { + fs: {}, + runtime: { + async exec() { + return { + id: "run-1", + async *[Symbol.asyncIterator]() { + yield { name: "stdout" as const, value: "started\n" }; + await new Promise(() => undefined); + }, + result: () => new Promise(() => undefined), + }; + }, + pending: async () => [], + detachExec: async () => true, + backends: () => [ + { id: "codemode", protocol: "module" as const, callable: true, detachable: true }, + ], + }, + }; + const tools = createTanStackTools({ + workspace: workspace as never, + exec: { detach: { afterMs: 150 } }, + streamEventName: "exec-progress", + format: "object", + }); + const events: Array> = []; + + const result = await tools.exec.execute({ command: "research()" } as never, { + toolCallId: "call-1", + emitCustomEvent: (_name, value) => events.push(value), + }); + + expect(result).toMatchObject({ executionId: "run-1", status: "running" }); + expect(events.length).toBeGreaterThanOrEqual(1); + expect(events.every((event) => !("status" in (event.snapshot as object)))).toBe(true); + }); + it("emits a running snapshot before the command produces more output", async () => { const quiet = quietCommandBackend(); const workspace = new Workspace({ diff --git a/packages/computer/tests/script-runner-worker.ts b/packages/computer/tests/script-runner-worker.ts index 2b5c6e76..16befd41 100644 --- a/packages/computer/tests/script-runner-worker.ts +++ b/packages/computer/tests/script-runner-worker.ts @@ -6,6 +6,7 @@ import { createGitClient } from "../src/git/index"; import type { WorkspaceExecutionMetadata, WorkspaceRuntimePendingCall, + WorkspaceRuntimeExecExit, WorkspaceRuntimeValue, WorkspaceStub, } from "../src/index"; @@ -23,6 +24,7 @@ import { createArtifactsModule } from "../src/modules/artifacts"; import { type BrowserModuleSession, createBrowserModule } from "../src/modules/browser"; import { createContainerModule } from "../src/modules/container"; import { createGitModule } from "../src/modules/git"; +import { defineExec, type ExecToolOutput } from "../src/tools/common/exec"; export interface Env { HOST: DurableObjectNamespace; @@ -75,6 +77,11 @@ function fakeContainerBackend(): WorkspaceBackend { const durableCalls = new Map(); // Each call onWait was told about. const waits: WorkspaceRuntimePendingCall[] = []; +// Each exit onDetachedExit was handed, as often as it was handed it. +const exits: WorkspaceRuntimeExecExit[] = []; +// Execs whose next onDetachedExit call throws, as a hook that couldn't +// reach its app would. +const refuseExit = new Set(); let gateOpen = false; // Named gates a run can wait on after the first, for a second restart. const opened = new Set(); @@ -301,13 +308,16 @@ const testStreams = streamsModule.module({ // the new module registry, so the same backend runs on both. const NEW_REGISTRY_BACKEND = "codemode-new-registry"; const NETWORK_BACKEND = "codemode-network"; +// A backend whose runs replay by default, as the exec tool's runs need +// to wait for an answer: the tool doesn't choose a run's policy. +const REPLAY_BACKEND = "codemode-replay"; function codemodeBackend( env: Env, overrides: Partial< Pick< CodemodeBackendOptions, - "id" | "compatibilityFlags" | "loader" | "egress" | "maxStdioBytes" + "id" | "compatibilityFlags" | "loader" | "egress" | "maxStdioBytes" | "onInterrupt" > > = {}, ) { @@ -317,6 +327,10 @@ function codemodeBackend( maxStdioBytes: overrides.maxStdioBytes ?? 64, maxCapabilityBytes: 1024, onWait: (call) => void waits.push(call), + onDetachedExit: (exec) => { + if (refuseExit.delete(exec.id)) throw new Error("the app is away"); + exits.push(exec); + }, maxConcurrentCapabilityCalls: 2, modules: { "math-kit": "export const double = (value) => value * 2;", @@ -450,6 +464,7 @@ export class HostDO extends DurableObject { env.LOADER.load({ ...code, allowExperimental: true }), }, }), + codemodeBackend(env, { id: REPLAY_BACKEND, onInterrupt: "replay" }), codemodeBackend(env, { id: NETWORK_BACKEND, egress: { mode: "http-gateway", gateway: ctx.exports.TestGateway }, @@ -534,6 +549,7 @@ export class HostDO extends DurableObject { onInterrupt?: "fail" | "restart" | "replay"; network?: boolean; metadata?: Record; + detach?: boolean; }) { gateOpen = false; opened.clear(); @@ -547,6 +563,7 @@ export class HostDO extends DurableObject { metadata: input.metadata, }); void handle.result().catch(() => undefined); + if (input.detach === true) await this.#workspace.runtime.detachExec(handle.id); // Let the run reach the gate. for (let attempt = 0; attempt < 500 && !durableCalls.has("gate"); attempt += 1) { await new Promise((resolve) => setTimeout(resolve, 10)); @@ -582,9 +599,39 @@ export class HostDO extends DurableObject { return waits; } + // Runs source through the exec tool with `detach`, as an agent would, + // and returns the tool's last reply. + async execDetached(source: string, metadata: Record) { + const exec = defineExec({ + workspace: this.#workspace, + backends: { [REPLAY_BACKEND]: {} }, + metadata, + detach: true, + }); + let reply: ExecToolOutput | undefined; + for await (const snapshot of exec.execute({ command: source })) reply = snapshot; + return reply; + } + + exits(id: string) { + return exits.filter((exec) => exec.id === id); + } + + refuseExit(id: string) { + refuseExit.add(id); + } + + async detachExec(id: string, backend?: string) { + return this.#workspace.runtime.detachExec(id, { backend }); + } + async respond(id: string, callId: string, value: WorkspaceRuntimeValue) { try { - await this.#workspace.runtime.respond(id, callId, value); + // Answer on the backend the call waits on, which pending() names. + const backend = (await this.#workspace.runtime.pending()).find( + (call) => call.id === id && call.callId === callId, + )?.backend; + await this.#workspace.runtime.respond(id, callId, value, { backend }); return { ok: true as const }; } catch (error) { return { ok: false as const, code: (error as { code?: string }).code }; @@ -870,6 +917,7 @@ export default class extends WorkerEntrypoint { onInterrupt?: "fail" | "restart" | "replay"; network?: boolean; metadata?: Record; + detach?: boolean; }, ), }); @@ -903,6 +951,33 @@ export default class extends WorkerEntrypoint { return Response.json(await stub.waits()); } + if (url.pathname === "/exec-detached") { + const body = (await request.json()) as { + source: string; + metadata: Record; + }; + return Response.json(await stub.execDetached(body.source, body.metadata)); + } + + if (url.pathname === "/exits") { + return Response.json(await stub.exits(url.searchParams.get("id") ?? "")); + } + + if (url.pathname === "/detach") { + const id = url.searchParams.get("id") ?? ""; + const backend = url.searchParams.get("backend") ?? undefined; + try { + return Response.json({ detached: await stub.detachExec(id, backend) }); + } catch (error) { + return Response.json({ error: String(error) }, { status: 400 }); + } + } + + if (url.pathname === "/refuse-exit") { + await stub.refuseExit(url.searchParams.get("id") ?? ""); + return new Response(null, { status: 204 }); + } + if (url.pathname === "/interrupt-pending") { return Response.json(await stub.pending()); } diff --git a/packages/computer/tests/script-runner.test.ts b/packages/computer/tests/script-runner.test.ts index 83a901fa..bfa1790b 100644 --- a/packages/computer/tests/script-runner.test.ts +++ b/packages/computer/tests/script-runner.test.ts @@ -1037,6 +1037,7 @@ describe("durable executions", () => { onInterrupt?: string; network?: boolean; metadata?: Record; + detach?: boolean; }, ) { const response = await SELF.fetch("https://example.test/durable-start", { @@ -1080,6 +1081,23 @@ describe("durable executions", () => { expect(await calls()).toEqual({ work: 2, gate: 2 }); }); + it("hands onDetachedExit the metadata of a detached run that failed because it was interrupted", async () => { + await start("exit-fail-metadata", { + source, + onInterrupt: "fail", + metadata: { conversation: "c2" }, + detach: true, + }); + expect(await result("exit-fail-metadata")).toMatchObject({ status: "failed" }); + let heard: Array<{ status: string; metadata: Record }> = []; + for (let attempt = 0; attempt < 200 && heard.length === 0; attempt += 1) { + const response = await SELF.fetch("https://example.test/exits?id=exit-fail-metadata"); + heard = (await response.json()) as typeof heard; + if (heard.length === 0) await new Promise((resolve) => setTimeout(resolve, 20)); + } + expect(heard).toMatchObject([{ status: "failed", metadata: { conversation: "c2" } }]); + }); + it("replays the clock a run read before it was interrupted", async () => { await start("durable-clock", { source: ` @@ -1971,6 +1989,133 @@ describe("interrupts", () => { // The test backend keeps 64 bytes of standard error. expect(settled.stderr).toContain("ctx.interrupt() needs a replayable run"); }); + + type Exit = { + id: string; + backend: string; + status: string; + exitCode: number; + result?: unknown; + metadata: Record; + stdout: string; + stderr: string; + exitedAt: number; + }; + + async function exits(id: string): Promise { + const response = await SELF.fetch(`https://example.test/exits?id=${encodeURIComponent(id)}`); + return (await response.json()) as Exit[]; + } + + // What onDetachedExit was handed for an exec, once it has anything. + async function exited(id: string): Promise { + for (let attempt = 0; attempt < 200; attempt += 1) { + const heard = await exits(id); + if (heard.length > 0) return heard; + await new Promise((resolve) => setTimeout(resolve, 20)); + } + throw new Error(`onDetachedExit never heard about ${id}`); + } + + async function detach(id: string): Promise<{ detached?: boolean; error?: string }> { + const response = await SELF.fetch(`https://example.test/detach?id=${encodeURIComponent(id)}`, { + method: "POST", + }); + return (await response.json()) as { detached?: boolean; error?: string }; + } + + it("hands onDetachedExit how a detached run ended, once, after its answer", async () => { + await start("exit-answer", approval, { metadata: { conversation: "c1" } }); + const call = await waitingCall("exit-answer"); + expect(await detach("exit-answer")).toEqual({ detached: true }); + expect(await exits("exit-answer")).toEqual([]); + await respond("exit-answer", call.callId, { ok: true }); + expect(await result("exit-answer")).toMatchObject({ status: "completed" }); + const [heard, ...rest] = await exited("exit-answer"); + expect(rest).toEqual([]); + expect(heard).toMatchObject({ + id: "exit-answer", + backend: "codemode", + status: "completed", + exitCode: 0, + result: { ok: true }, + metadata: { conversation: "c1" }, + }); + expect(heard?.stdout).toContain("after true"); + expect(heard?.exitedAt).toBeGreaterThanOrEqual(call.since); + }); + + it("hands onDetachedExit nothing for a run nothing detached", async () => { + await start("exit-attached", approval); + const call = await waitingCall("exit-attached"); + await respond("exit-attached", call.callId, { ok: true }); + expect(await result("exit-attached")).toMatchObject({ status: "completed" }); + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(await exits("exit-attached")).toEqual([]); + expect(await detach("exit-attached")).toEqual({ detached: false }); + }); + + it("hands onDetachedExit a detached run killed while it waits", async () => { + await start("exit-kill", approval); + await waitingCall("exit-kill"); + await detach("exit-kill"); + await SELF.fetch("https://example.test/runtime-kill?id=exit-kill", { method: "POST" }); + expect(await exited("exit-kill")).toMatchObject([ + { id: "exit-kill", status: "cancelled", exitCode: 130 }, + ]); + }); + + it("hands onDetachedExit an exit again from recover() when the hook threw", async () => { + await SELF.fetch("https://example.test/refuse-exit?id=exit-retry", { method: "POST" }); + await start("exit-retry", approval); + const call = await waitingCall("exit-retry"); + await detach("exit-retry"); + await respond("exit-retry", call.callId, { ok: true }); + expect(await result("exit-retry")).toMatchObject({ status: "completed" }); + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(await exits("exit-retry")).toEqual([]); + await SELF.fetch("https://example.test/durable-recover", { method: "POST" }); + expect(await exits("exit-retry")).toMatchObject([{ status: "completed" }]); + }); + + it("lets the exec tool detach from a run once it waits, and hears how it ended later", async () => { + const response = await SELF.fetch("https://example.test/exec-detached", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ source: approval, metadata: { conversation: "c3" } }), + }); + const reply = (await response.json()) as { + executionId: string; + status: string; + waiting: Array<{ callId: string; payload: unknown }>; + }; + expect(reply).toMatchObject({ + backend: "codemode-replay", + status: "waiting", + waiting: [{ payload: { kind: "approval", action: "deploy" } }], + }); + // The tool came back, and the run waits on without it. + const call = await waitingCall(reply.executionId); + expect(call).toMatchObject({ backend: "codemode-replay", metadata: { conversation: "c3" } }); + expect(await exits(reply.executionId)).toEqual([]); + await respond(reply.executionId, call.callId, { ok: true }); + expect(await exited(reply.executionId)).toMatchObject([ + { status: "completed", result: { ok: true }, metadata: { conversation: "c3" } }, + ]); + }); + + it("hands onDetachedExit an exit after a restart when the hook missed it", async () => { + await SELF.fetch("https://example.test/refuse-exit?id=exit-restart", { method: "POST" }); + await start("exit-restart", approval); + const call = await waitingCall("exit-restart"); + await detach("exit-restart"); + await respond("exit-restart", call.callId, { ok: true }); + expect(await result("exit-restart")).toMatchObject({ status: "completed" }); + await SELF.fetch("https://example.test/durable-restart", { method: "POST" }); + // Any call reconnects the backend, which delivers what it still holds. + await pending("exit-restart"); + expect(await exited("exit-restart")).toMatchObject([{ status: "completed" }]); + }); }); describe("codemode module objects", () => {