From ea10696b44f10dba9702b68942d89f12ce21a881 Mon Sep 17 00:00:00 2001 From: scuffi Date: Mon, 19 Oct 2026 15:16:42 +0100 Subject: [PATCH 01/12] computer: Let host module calls wait for an answer A host module function can now call ctx.interrupt(payload) to pause a replayable run until someone answers, for example to approve an action the run is about to take. The code being run just calls the function and gets its result, however long the answer takes. The interrupted call is journaled as waiting. The bridge holds new calls, lets the ones in flight finish and be journaled, and the backend then releases the isolate while the run stays running. Waiting runs are left alone by recovery, so they survive a deploy, and their wait doesn't count toward the timeout. workspace.runtime.pending() lists the waiting calls and respond() answers one. Once none are waiting, the run replays to the call, which runs again with the same callId and gets its answer from interrupt() this time. Only a replayable run can wait, since a pause needs the journal, and interrupt() throws elsewhere. Calls that interrupt after the run starts releasing wait once it resumes, so approvals made together are asked one after another. A released attempt's late output is dropped, as it isn't the run's. --- .changeset/js-exec-interrupts.md | 5 + examples/rlm/worker/capability.test.ts | 1 + .../worker-javascript/journal.test.ts | 37 ++++ .../src/backends/worker-javascript/journal.ts | 168 +++++++++++++++- .../worker-javascript/worker-javascript.ts | 158 +++++++++++++--- packages/computer/src/index.ts | 1 + .../computer/src/modules/artifacts.test.ts | 1 + packages/computer/src/modules/assets.test.ts | 1 + .../computer/src/modules/container.test.ts | 1 + packages/computer/src/modules/tools.test.ts | 1 + packages/computer/src/runtime/bridge.test.ts | 13 ++ packages/computer/src/runtime/bridge.ts | 73 ++++++- packages/computer/src/runtime/runtime.ts | 38 ++++ packages/computer/src/runtime/types.ts | 37 ++++ .../src/tools/ai-sdk/bindings.test.ts | 1 + .../computer/src/tools/pi-ai/bindings.test.ts | 1 + .../computer/tests/script-runner-worker.ts | 77 ++++++++ packages/computer/tests/script-runner.test.ts | 179 ++++++++++++++++++ 18 files changed, 756 insertions(+), 37 deletions(-) create mode 100644 .changeset/js-exec-interrupts.md diff --git a/.changeset/js-exec-interrupts.md b/.changeset/js-exec-interrupts.md new file mode 100644 index 000000000..7a2a049b3 --- /dev/null +++ b/.changeset/js-exec-interrupts.md @@ -0,0 +1,5 @@ +--- +"@cloudflare/computer": minor +--- + +Add experimental `ctx.interrupt()` for host module functions, with `runtime.pending()` and `runtime.respond()`, so a replayable run can wait for an answer such as an approval. diff --git a/examples/rlm/worker/capability.test.ts b/examples/rlm/worker/capability.test.ts index b17848ebb..6aaf5f084 100644 --- a/examples/rlm/worker/capability.test.ts +++ b/examples/rlm/worker/capability.test.ts @@ -40,6 +40,7 @@ function context(signal = new AbortController().signal): WorkspaceModuleCallCont deadline: Date.now() + 1_000, access: "read", callId: "test-call", + interrupt: () => Promise.reject(new Error("not used")), resolvePath: async (path) => path, }; } diff --git a/packages/computer/src/backends/worker-javascript/journal.test.ts b/packages/computer/src/backends/worker-javascript/journal.test.ts index 2596e2d4a..cb796f621 100644 --- a/packages/computer/src/backends/worker-javascript/journal.test.ts +++ b/packages/computer/src/backends/worker-javascript/journal.test.ts @@ -3,12 +3,14 @@ import { SQLiteTestStorage } from "@cloudflare/dofs/testing"; import { describe, expect, it, vi } from "vitest"; import { + answerCall, callKey, createDurableTables, decodeValue, describeCall, encodeValue, ExecutionJournal, + waitingCalls, } from "./journal.js"; function database() { @@ -97,6 +99,41 @@ describe("ExecutionJournal", () => { expect(replay.expectedCall()).toBeUndefined(); }); + it("holds an interrupted call until it's answered, then gives it the answer", () => { + const db = database(); + db.run( + `INSERT INTO workspace_runtime_durable (backend, execution_id, policy, attempt) + VALUES ('js', 'run', 'replay', 1)`, + ); + const first = new ExecutionJournal(db, "js", "run"); + const step = first.begin("host/ws:test.approve", ["deploy"]); + if (step.kind !== "live") throw new Error("expected a live call"); + first.interrupt(step.seq, { action: "deploy" }); + expect(waitingCalls(db, "js")).toEqual([ + { id: "run", seq: 1, payload: { action: "deploy" }, since: expect.any(Number) }, + ]); + expect(answerCall(db, "js", "run", 2, true)).toBeUndefined(); + expect(answerCall(db, "js", "run", 1, { ok: true })).toBe(0); + expect(answerCall(db, "js", "run", 1, { ok: true })).toBeUndefined(); + expect(waitingCalls(db, "js")).toEqual([]); + const answer = { value: { ok: true } }; + const resumed = new ExecutionJournal(db, "js", "run"); + expect(resumed.begin("host/ws:test.approve", ["deploy"])).toEqual({ + kind: "live", + seq: 1, + rerun: false, + answer, + }); + // Interrupted again while it runs with its answer: a rerun that keeps it. + const again = new ExecutionJournal(db, "js", "run"); + expect(again.begin("host/ws:test.approve", ["deploy"])).toEqual({ + kind: "live", + seq: 1, + rerun: true, + answer, + }); + }); + it("numbers responses in the order they were committed", () => { const db = database(); const first = new ExecutionJournal(db, "js", "run"); diff --git a/packages/computer/src/backends/worker-javascript/journal.ts b/packages/computer/src/backends/worker-javascript/journal.ts index 202bdf527..bad9f5a4f 100644 --- a/packages/computer/src/backends/worker-javascript/journal.ts +++ b/packages/computer/src/backends/worker-javascript/journal.ts @@ -65,14 +65,29 @@ export interface DurableRecord { readonly policy: Exclude; readonly attempt: number; readonly start: DurableStart; + /** When the run's first unanswered interrupt was made, if one is waiting. */ + readonly waitingSince: number | null; + /** Time the run has spent waiting for answers, which its timeout leaves out. */ + readonly waitedMs: number; +} + +/** A call waiting for an answer, as `pending()` lists it. */ +export interface WaitingCall { + readonly id: string; + readonly seq: number; + readonly payload: unknown; + readonly since: number; } interface JournalEntry { readonly seq: number; readonly key: string; readonly label: string; - readonly state: "executing" | "applied"; + // An interrupted call is `waiting` until it's answered, then `answered` + // until the resumed run makes it again. + readonly state: "executing" | "applied" | "waiting" | "answered"; readonly response: string | null; + readonly answer: string | null; readonly at: number | null; readonly ordinal: number | null; } @@ -90,6 +105,8 @@ export type JournalStep = readonly seq: number; /** The call was in flight when the run was interrupted. */ readonly rerun: boolean; + /** The answer to the call's interrupt, which it now returns. */ + readonly answer?: { readonly value: unknown }; } | { readonly kind: "diverge"; readonly message: string }; @@ -115,6 +132,8 @@ export function createDurableTables(db: Database): void { execution_id TEXT NOT NULL, policy TEXT NOT NULL, attempt INTEGER NOT NULL, + waiting_since INTEGER, + waited_ms INTEGER NOT NULL DEFAULT 0, PRIMARY KEY (backend, execution_id) ) `); @@ -136,6 +155,8 @@ export function createDurableTables(db: Database): void { label TEXT NOT NULL, state TEXT NOT NULL, response TEXT, + payload TEXT, + answer TEXT, at INTEGER, ordinal INTEGER, PRIMARY KEY (backend, execution_id, seq) @@ -193,8 +214,14 @@ export function saveDurableStart( } export function loadDurable(db: Database, backend: string, id: string): DurableRecord | undefined { - const row = db.one<{ policy: string; attempt: number }>( - `SELECT policy, attempt FROM workspace_runtime_durable WHERE backend = ? AND execution_id = ?`, + const row = db.one<{ + policy: string; + attempt: number; + waiting_since: number | null; + waited_ms: number; + }>( + `SELECT policy, attempt, waiting_since, waited_ms FROM workspace_runtime_durable + WHERE backend = ? AND execution_id = ?`, backend, id, ); @@ -207,7 +234,13 @@ export function loadDurable(db: Database, backend: string, id: string): DurableR ); try { const start = JSON.parse(chunks.map((chunk) => chunk.data).join("")) as DurableStart; - return { policy: row.policy, attempt: Number(row.attempt), start }; + return { + policy: row.policy, + attempt: Number(row.attempt), + start, + waitingSince: row.waiting_since === null ? null : Number(row.waiting_since), + waitedMs: Number(row.waited_ms), + }; } catch { return undefined; } @@ -222,6 +255,69 @@ export function recordAttempt(db: Database, backend: string, id: string, attempt ); } +/** + * Answer a waiting call. Returns how many of the run's calls are still + * waiting, or `undefined` if this one wasn't. Once none are, the run's + * wait ends and its timeout counts again. + */ +export function answerCall( + db: Database, + backend: string, + id: string, + seq: number, + value: unknown, +): number | undefined { + const answered = db.all( + `UPDATE workspace_runtime_journal SET state = 'answered', answer = ? + WHERE backend = ? AND execution_id = ? AND seq = ? AND state = 'waiting' + RETURNING seq`, + encodeLimited(value, "answer"), + backend, + id, + seq, + ); + if (answered.length === 0) return undefined; + const waiting = Number( + db.one<{ count: number }>( + `SELECT COUNT(*) AS count FROM workspace_runtime_journal + WHERE backend = ? AND execution_id = ? AND state = 'waiting'`, + backend, + id, + )?.count ?? 0, + ); + if (waiting === 0) { + db.run( + `UPDATE workspace_runtime_durable + SET waited_ms = waited_ms + MAX(0, ? - waiting_since), waiting_since = NULL + WHERE backend = ? AND execution_id = ? AND waiting_since IS NOT NULL`, + Date.now(), + backend, + id, + ); + } + return waiting; +} + +/** Every call of this backend's runs that's waiting for an answer. */ +export function waitingCalls(db: Database, backend: string): WaitingCall[] { + return db + .all<{ execution_id: string; seq: number; payload: string; waiting_since: number }>( + `SELECT j.execution_id, j.seq, j.payload, d.waiting_since + FROM workspace_runtime_journal j + JOIN workspace_runtime_durable d + ON d.backend = j.backend AND d.execution_id = j.execution_id + WHERE j.backend = ? AND j.state = 'waiting' + ORDER BY d.waiting_since, j.execution_id, j.seq`, + backend, + ) + .map((row) => ({ + id: row.execution_id, + seq: Number(row.seq), + payload: decodeValue(row.payload), + since: Number(row.waiting_since), + })); +} + export function clearJournal(db: Database, backend: string, id: string): void { for (const table of [ "workspace_runtime_journal", @@ -283,10 +379,12 @@ export class ExecutionJournal implements BridgeJournal { label: string; state: string; response: string | null; + answer: string | null; at: number | null; ordinal: number | null; }>( - `SELECT seq, call_key, label, state, response, at, ordinal FROM workspace_runtime_journal + `SELECT seq, call_key, label, state, response, answer, at, ordinal + FROM workspace_runtime_journal WHERE backend = ? AND execution_id = ? ORDER BY seq`, backend, id, @@ -295,9 +393,10 @@ export class ExecutionJournal implements BridgeJournal { seq: Number(row.seq), key: row.call_key, label: row.label, - state: row.state === "applied" && row.response !== null ? "applied" : "executing", + state: journalState(row.state, row.response), // An empty response marks one stored in chunks. response: row.response === "" ? this.#chunkedResponse(Number(row.seq)) : row.response, + answer: row.answer, at: row.at === null ? null : Number(row.at), ordinal: row.ordinal === null ? null : Number(row.ordinal), })); @@ -399,9 +498,23 @@ export class ExecutionJournal implements BridgeJournal { ordinal: entry.ordinal ?? 0, }; } + const answer = + entry.answer === null ? undefined : { value: decodeValue(entry.answer) as unknown }; + // An answered call runs again by design, so it isn't a rerun until + // it's under way. + if (entry.state === "answered") { + this.#db.run( + `UPDATE workspace_runtime_journal SET state = 'executing' + WHERE backend = ? AND execution_id = ? AND seq = ?`, + this.#backend, + this.#id, + entry.seq, + ); + return { kind: "live", seq: entry.seq, rerun: false, answer }; + } // The call was in flight when the execution was interrupted, so its // response never reached the code. Run it again. - return { kind: "live", seq: entry.seq, rerun: true }; + return { kind: "live", seq: entry.seq, rerun: true, answer }; } if (this.#unanswered > 0) { return { @@ -458,6 +571,29 @@ export class ExecutionJournal implements BridgeJournal { return { at, ordinal: this.#lastOrdinal }; } + /** + * Mark a live call as waiting for an answer. Its response is never + * committed: the run is released, and the call runs again once + * answered. + */ + interrupt(seq: number, payload: unknown): void { + this.#db.run( + `UPDATE workspace_runtime_journal SET state = 'waiting', payload = ? + WHERE backend = ? AND execution_id = ? AND seq = ?`, + encodeLimited(payload, "payload"), + this.#backend, + this.#id, + seq, + ); + this.#db.run( + `UPDATE workspace_runtime_durable SET waiting_since = COALESCE(waiting_since, ?) + WHERE backend = ? AND execution_id = ?`, + Date.now(), + this.#backend, + this.#id, + ); + } + diverged(message: string): void { this.#onDiverge(message); } @@ -537,6 +673,24 @@ function cyrb53(text: string, seed: number): string { return (4294967296 * (2097151 & h2) + (h1 >>> 0)).toString(36).padStart(11, "0"); } +function journalState(state: string, response: string | null): JournalEntry["state"] { + if (state === "applied") return response === null ? "executing" : "applied"; + return state === "waiting" || state === "answered" ? state : "executing"; +} + +// Interrupt payloads and answers are stored in one row each. +const MAX_WAIT_CHARACTERS = CHUNK_CHARACTERS; + +function encodeLimited(value: unknown, what: string): string { + const text = encodeValue(value); + if (text.length > MAX_WAIT_CHARACTERS) { + throw new Error( + `An interrupt's ${what} must encode to at most ${MAX_WAIT_CHARACTERS} characters.`, + ); + } + return text; +} + const BYTES_TAG = "\u0000bytes"; export function encodeValue(value: unknown): string { diff --git a/packages/computer/src/backends/worker-javascript/worker-javascript.ts b/packages/computer/src/backends/worker-javascript/worker-javascript.ts index 572c3eabc..67beb9afb 100644 --- a/packages/computer/src/backends/worker-javascript/worker-javascript.ts +++ b/packages/computer/src/backends/worker-javascript/worker-javascript.ts @@ -18,10 +18,12 @@ import type { WorkspaceRuntimeEvent, WorkspaceRuntimeInterruptPolicy, WorkspaceRuntimeLoader, + WorkspaceRuntimePendingCall, WorkspaceRuntimeValue, } from "../../runtime/types.js"; import { decodeRuntimeFrames, type RuntimeFrame } from "./frames.js"; import { + answerCall, clearDurable, createDurableTables, DURABLE_TABLES, @@ -36,6 +38,7 @@ import { recordAttempt, saveDurableStart, toBase64, + waitingCalls, } from "./journal.js"; import { assertHostModuleExports, @@ -253,6 +256,9 @@ interface ExecutionRecord { // Output a replayed run already recorded. Its frames must repeat these // bytes, which the record checks and doesn't append again. replayed?: Record<"stdout" | "stderr", { bytes: Uint8Array; offset: number }>; + // Set while a run's isolate is being released for an interrupt, and + // once it is, until the run resumes. + waiting?: "releasing" | "released"; } export class WorkerJavaScriptBackend implements WorkspaceModuleBackend { @@ -486,45 +492,132 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { const record = this.#loadRecord(id); if (record?.status !== "running") return; this.#records.set(id, record); - const fail = (message: string) => - this.#finish(record, "failed", [ - { - id, - seq: record.events.length + 1, - name: "stderr", - value: new TextEncoder().encode(`${message}\n`), - }, - { id, seq: record.events.length + 2, name: "exit", code: 1 }, - ]); + const saved = this.#loadDurable(record); + if (saved === undefined) return; + // A run waiting for an answer resumes once it has every one. + if (saved.waitingSince !== null) { + record.waiting = "released"; + return; + } + if (saved.attempt >= MAX_INTERRUPT_ATTEMPTS) { + this.#failInterrupted( + record, + `Execution was interrupted ${saved.attempt} times and won't be started again.`, + ); + return; + } + this.#queueRelaunch(record, saved, true); + } + + #loadDurable(record: ExecutionRecord) { let saved: ReturnType; try { - saved = loadDurable(this.#host.db, this.#backendId, id); + saved = loadDurable(this.#host.db, this.#backendId, record.id); } catch { saved = undefined; } if (saved === undefined) { - fail("Execution was interrupted when its Workspace runtime restarted."); - return; - } - const { attempt, start } = saved; - if (attempt >= MAX_INTERRUPT_ATTEMPTS) { - fail(`Execution was interrupted ${attempt} times and won't be started again.`); - return; + this.#failInterrupted( + record, + "Execution was interrupted when its Workspace runtime restarted.", + ); } + return saved; + } + + #failInterrupted(record: ExecutionRecord, message: string) { + this.#finish(record, "failed", [ + { + id: record.id, + seq: record.events.length + 1, + name: "stderr", + value: new TextEncoder().encode(`${message}\n`), + }, + { id: record.id, seq: record.events.length + 2, name: "exit", code: 1 }, + ]); + } + + // Queue a run to start again, on the runtime code it started on. + // Resuming after an answer isn't an attempt. + #queueRelaunch( + record: ExecutionRecord, + saved: NonNullable>, + counts: boolean, + ) { + const { start } = saved; if (start.version !== JOURNAL_VERSION || start.fingerprint !== this.#fingerprint(start)) { - fail( + this.#failInterrupted( + record, "Execution was interrupted and can't resume, because the JavaScript runtime changed since it started.", ); return; } - const resumable = saved; this.#recoveryQueue.push(() => { if (record.status !== "running" || record.finalization) return; - this.#recovering.add(id); - this.#relaunch(record, resumable, fail); + this.#recovering.add(record.id); + this.#relaunch(record, saved, counts); }); } + // A call interrupted to wait for an answer. Let the run's other calls + // finish and be journaled, then release its isolate. The run stays + // running and resumes once every waiting call is answered. + async #release(record: ExecutionRecord) { + const { bridge } = record; + if (record.waiting !== undefined || bridge === undefined) return; + record.waiting = "releasing"; + await bridge.quiesce(); + // It may have settled meanwhile, for example by failing on its own. + if (record.status !== "running" || record.finalization) return; + record.bridge = undefined; + record.control?.cancel(); + await record.control?.completion.catch(() => undefined); + await bridge.cancelAndDrain().catch(() => undefined); + record.control = undefined; + record.journal = undefined; + record.waiting = "released"; + this.#markSettled(record.id); + this.#resumeIfAnswered(record); + } + + #resumeIfAnswered(record: ExecutionRecord) { + if (record.status !== "running" || record.waiting !== "released") return; + const saved = this.#loadDurable(record); + if (saved === undefined || saved.waitingSince !== null) return; + this.#queueRelaunch(record, saved, false); + this.#drainRecoveries(); + } + + async pending(): Promise[]> { + return waitingCalls(this.#host.db, this.#backendId).map((call) => ({ + id: call.id, + callId: `${call.id}:${call.seq}`, + payload: call.payload as WorkspaceRuntimeValue, + since: call.since, + })); + } + + async respond(input: { id: string; callId: string; value: WorkspaceRuntimeValue }) { + assertExecutionId(input.id); + assertRuntimeValue(input.value); + const record = this.#recordFor(input.id); + const prefix = `${input.id}:`; + const seq = input.callId.startsWith(prefix) ? Number(input.callId.slice(prefix.length)) : NaN; + const waiting = + record?.status === "running" && Number.isSafeInteger(seq) + ? this.#host.db.transactionSync(() => + answerCall(this.#host.db, this.#backendId, input.id, seq, input.value), + ) + : undefined; + if (record === undefined || waiting === undefined) { + throw runtimeError( + "ENOTWAITING", + `execution ${input.id} has no call ${input.callId} waiting for an answer`, + ); + } + if (waiting === 0) this.#resumeIfAnswered(record); + } + // A Durable Object can run only so many Dynamic Workers at once, and // new executions need some of them, so interrupted runs start a few at // a time. Each one that settles lets the next start. @@ -538,19 +631,22 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { #relaunch( record: ExecutionRecord, - { policy, attempt, start }: NonNullable>, - fail: (message: string) => void, + { policy, attempt, start, waitedMs }: NonNullable>, + counts: boolean, ) { const id = record.id; + const fail = (message: string) => this.#failInterrupted(record, message); // A run's timeout counts from when it first started, including the - // time it spent interrupted or waiting to resume. - const remainingMs = start.startedAt + start.timeoutMs - Date.now(); + // time it spent interrupted or waiting to resume, but not the time it + // waited for answers. + const remainingMs = start.startedAt + start.timeoutMs + waitedMs - Date.now(); if (remainingMs <= 0) { fail("Execution was interrupted and its timeout passed before it could resume."); return; } + record.waiting = undefined; try { - recordAttempt(this.#host.db, this.#backendId, id, attempt + 1); + if (counts) recordAttempt(this.#host.db, this.#backendId, id, attempt + 1); let journal: ExecutionJournal | undefined; let durable: WorkspaceExecutionContext["durable"]; if (policy === "replay") { @@ -799,6 +895,7 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { maxResultBytes: this.#options.maxResultBytes, onAttachOutput: (readable) => this.#pumpFrames(record, readable), callIdPrefix: record.id, + onWait: () => void this.#release(record), ...hostEgressFetch(this.#options.egress), ...(spec.journal === undefined ? {} : { journal: spec.journal }), }); @@ -1011,6 +1108,9 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { // as it arrives. Resolves when the stream closes, which is the // signal the runner's evaluate awaits before returning. async #pumpFrames(record: ExecutionRecord, readable: ReadableStream) { + // A run released for an interrupt has moved on from this attempt, and + // whatever its isolate writes on the way out isn't the run's output. + const attempt = record.bridge; const reader = decodeRuntimeFrames(readable).getReader(); try { while (true) { @@ -1031,7 +1131,7 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { if (record.status !== "running") break; throw error; } - if (next.done) break; + if (next.done || record.bridge !== attempt) break; this.#ingestFrame(record, next.value); } } finally { @@ -1090,6 +1190,8 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { #finalize(record: ExecutionRecord, errorMessage?: string): Promise { if (record.finalization) return record.finalization; if (record.status !== "running") return Promise.resolve(); + // Its isolate ending for an interrupt isn't the run ending. + if (record.waiting === "released") return Promise.resolve(); const finalization = this.#finalizeOnce(record, errorMessage); record.finalization = finalization; return finalization; diff --git a/packages/computer/src/index.ts b/packages/computer/src/index.ts index f8bd47dd2..cdbd71f91 100644 --- a/packages/computer/src/index.ts +++ b/packages/computer/src/index.ts @@ -108,6 +108,7 @@ export type { WorkspaceRuntimeKillOptions, WorkspaceRuntimeLoader, WorkspaceRuntimeOutputOptions, + WorkspaceRuntimePendingCall, WorkspaceRuntimeResult, WorkspaceRuntimeStatus, WorkspaceRuntimeTruncation, diff --git a/packages/computer/src/modules/artifacts.test.ts b/packages/computer/src/modules/artifacts.test.ts index b22e36c09..ec2b75423 100644 --- a/packages/computer/src/modules/artifacts.test.ts +++ b/packages/computer/src/modules/artifacts.test.ts @@ -43,6 +43,7 @@ function callContext(access: "read" | "read-write" = "read-write"): WorkspaceMod deadline: Date.now() + 60_000, access, callId: "test-call", + 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 dd2f88b5d..eb564ea5f 100644 --- a/packages/computer/src/modules/assets.test.ts +++ b/packages/computer/src/modules/assets.test.ts @@ -41,6 +41,7 @@ function callContext( deadline: Date.now() + 60_000, access: "read-write", callId: "test-call", + 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 e9947e3e0..1ab720520 100644 --- a/packages/computer/src/modules/container.test.ts +++ b/packages/computer/src/modules/container.test.ts @@ -90,6 +90,7 @@ function callContext( deadline: Date.now() + 60_000, access: "read-write", callId: "test-call", + 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 6ccf4d4a6..f7db6e3b8 100644 --- a/packages/computer/src/modules/tools.test.ts +++ b/packages/computer/src/modules/tools.test.ts @@ -18,6 +18,7 @@ function callContext( deadline: Date.now() + 60_000, access: "read-write", callId: "test-call", + 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 c74f1ec5c..b6b4f787a 100644 --- a/packages/computer/src/runtime/bridge.test.ts +++ b/packages/computer/src/runtime/bridge.test.ts @@ -263,6 +263,7 @@ describe("WorkspaceRuntimeBridge journal", () => { if (!Array.isArray(events)) throw new Error("malformed event log"); logs.push([position, events]); }, + interrupt: () => undefined, diverged: (message) => diverged.push(message), stalled: () => diverged.push("stalled"), }; @@ -349,6 +350,18 @@ describe("WorkspaceRuntimeBridge journal", () => { expect(logs).toEqual([[0, [-1, -2]]]); }); + it("runs an unsafe call again once answered, but not if it was cut off after", async () => { + const { target, diverged } = journaled([ + { kind: "live", seq: 1, rerun: false, answer: { value: true } }, + { kind: "live", seq: 1, rerun: true, answer: { value: true } }, + ]); + await expect(target.call("host/ws:test.charge", args)).resolves.toMatchObject({ result: "c" }); + await expect(message(target.call("host/ws:test.charge", args))).resolves.toContain( + "isn't safe to repeat", + ); + expect(diverged).toHaveLength(1); + }); + it("passes a stalled replay to the journal", () => { const { target, diverged } = journaled([]); target.reportStall(); diff --git a/packages/computer/src/runtime/bridge.ts b/packages/computer/src/runtime/bridge.ts index 84a331a40..051d6b4c0 100644 --- a/packages/computer/src/runtime/bridge.ts +++ b/packages/computer/src/runtime/bridge.ts @@ -28,6 +28,13 @@ export class WorkspaceRuntimeBridge extends RpcTarget { #fetchesRunning = 0; readonly #fetchQueue: Array<() => void> = []; readonly #abortControllers = new Set(); + // Responses on their way to the isolate, by their call's controller. + readonly #responses = new Map, AbortController>(); + // Calls that interrupted, by their controller. + readonly #waiting = new Set(); + readonly #onWait?: () => void; + // Set once a call interrupts: new calls are held, never started. + #holding = false; #cancelled = false; #callTimedOut = false; #calls = 0; @@ -52,6 +59,8 @@ export class WorkspaceRuntimeBridge extends RpcTarget { fetch?: (request: Request) => Promise; // Prefix of every call's callId, such as the execution id. callIdPrefix?: string; + // Called when a call interrupts, to release the run. + onWait?: () => void; } = {}, ) { super(); @@ -68,6 +77,7 @@ export class WorkspaceRuntimeBridge extends RpcTarget { this.#journal = integrations.journal; this.#fetch = integrations.fetch; this.#callIdPrefix = integrations.callIdPrefix ?? crypto.randomUUID(); + this.#onWait = integrations.onWait; } // Drain the runner's framed output stream. The isolate passes the @@ -112,6 +122,9 @@ export class WorkspaceRuntimeBridge extends RpcTarget { return Promise.resolve(boundedError(error, this.#maxPayloadBytes)); } } + // The run is being released for an interrupt. A call it makes now + // runs live when it resumes. + if (this.#holding) return new Promise(() => undefined); let requestBytes: number; try { requestBytes = measureValue(args, this.#maxPayloadBytes, "request"); @@ -187,6 +200,7 @@ export class WorkspaceRuntimeBridge extends RpcTarget { signal: abort.signal, deadline, callId, + interrupt: (payload) => this.#interrupt(step, abort, payload), access: this.#capability.access, resolvePath: (path, options) => this.#capability.resolveConfined(path, options?.allowMissing ?? false), @@ -258,6 +272,8 @@ export class WorkspaceRuntimeBridge extends RpcTarget { } }, this.#maxPayloadBytes); const call = withDeadline(operation, this.#maxCallDurationMs, this.#maxPayloadBytes, () => { + // A call waiting for an answer isn't late: the run is released. + if (this.#waiting.has(abort)) return; this.#callTimedOut = true; abort.abort(new Error("Workspace capability call timed out.")); }); @@ -274,7 +290,7 @@ export class WorkspaceRuntimeBridge extends RpcTarget { // Errors count against the response budget too. The budget error // itself is short and not counted, and `maxCalls` bounds how many // the isolate can see. - return call + const delivered = call .then((response) => { if (isFetch) return "result" in response ? { result: response.result } : response; const bytes = "result" in response ? response.bytes : errorBytes(response); @@ -293,6 +309,8 @@ export class WorkspaceRuntimeBridge extends RpcTarget { return "result" in response ? { result: response.result } : response; }) .then((response) => { + // An interrupted call answers nothing: it runs again once answered. + if (this.#waiting.has(abort)) return new Promise(() => undefined); // Write before deliver: the journal holds the response before the // isolate can act on it. A filesystem change is synchronous SQL on // the same database, with no I/O between it and here, so it and @@ -300,6 +318,50 @@ export class WorkspaceRuntimeBridge extends RpcTarget { if (step?.kind !== "live" || this.#journal === undefined) return response; return { ...response, ...this.#journal.commit(step.seq, response) }; }); + this.#responses.set(delivered, abort); + void delivered.finally(() => this.#responses.delete(delivered)).catch(() => undefined); + return delivered; + } + + #interrupt( + step: ReturnType | undefined, + abort: AbortController, + payload: unknown, + ): Promise { + if (step?.kind !== "live" || this.#journal === undefined) { + return Promise.reject( + new Error( + 'ctx.interrupt() needs a replayable run. Run the execution with onInterrupt: "replay", or make that the backend\'s default.', + ), + ); + } + if (step.answer !== undefined) return Promise.resolve(step.answer.value as never); + if (this.#cancelled) + return Promise.reject(new Error("Workspace execution is being cancelled.")); + if (!this.#waiting.has(abort)) { + try { + this.#journal.interrupt(step.seq, payload); + } catch (error) { + return Promise.reject(error); + } + this.#waiting.add(abort); + this.#onWait?.(); + } + // Settles only when the run is released and the call aborted. + return new Promise((_, reject) => { + abort.signal.addEventListener("abort", () => reject(abort.signal.reason), { once: true }); + }); + } + + /** + * Hold new calls and wait for the ones in flight to be answered and + * journaled, apart from those that interrupted. The run can then be + * released without any call running twice when it resumes. + */ + async quiesce(): Promise { + this.#holding = true; + const answering = [...this.#responses].filter(([, abort]) => !this.#waiting.has(abort)); + await Promise.allSettled(answering.map(([response]) => response)); } // The isolate is waiting for a journaled call its replay never made. @@ -539,10 +601,17 @@ export interface BridgeJournal { readonly at: number; readonly ordinal: number; } - | { readonly kind: "live"; readonly seq: number; readonly rerun: boolean } + | { + readonly kind: "live"; + readonly seq: number; + readonly rerun: boolean; + readonly answer?: { readonly value: unknown }; + } | { readonly kind: "diverge"; readonly message: string }; /** Record a live call's response, and return its time and ordinal. */ commit(seq: number, response: BridgeResponse): { at: number; ordinal: number }; + /** Mark a live call as waiting for an answer. */ + interrupt(seq: number, payload: unknown): void; /** Save a chunk of the isolate's event log. Throws on a malformed one. */ record(position: unknown, events: unknown): void; /** Called when a call doesn't match the journal. */ diff --git a/packages/computer/src/runtime/runtime.ts b/packages/computer/src/runtime/runtime.ts index ef4acb5b5..ecd5c93ec 100644 --- a/packages/computer/src/runtime/runtime.ts +++ b/packages/computer/src/runtime/runtime.ts @@ -14,7 +14,9 @@ import { type WorkspaceRuntimeExecOptions, type WorkspaceRuntimeGetOptions, type WorkspaceRuntimeKillOptions, + type WorkspaceRuntimePendingCall, type WorkspaceRuntimeResult, + type WorkspaceRuntimeValue, } from "./types.js"; interface WorkspaceRuntimeRouterOptions { @@ -172,6 +174,42 @@ export class WorkspaceRuntime { } } + /** + * Host module calls waiting for an answer, from `ctx.interrupt()`, + * across every backend that supports it, oldest wait first. + */ + async pending(): Promise { + const calls: WorkspaceRuntimePendingCall[] = []; + for (const [id, backend] of this.#options.backends) { + if (!isModuleBackend(backend)) continue; + const handle = await this.#options.backendHandle(id); + for (const call of (await handle.pending?.()) ?? []) calls.push({ ...call, backend: id }); + } + return calls.sort((a, b) => a.since - b.since); + } + + /** + * Answer a call waiting in `ctx.interrupt()`. The run resumes once + * none of its calls are still waiting, and the call gets `value` as + * what `interrupt()` returns. Throws `ENOTWAITING` if the call isn't + * waiting, for example because it was already answered or the run + * was killed. + */ + async respond( + id: string, + callId: string, + value: WorkspaceRuntimeValue, + options: { backend?: string } = {}, + ): Promise { + assertExecutionId(id); + const backend = this.#backend(options.backend); + const handle = await this.#options.backendHandle(backend); + if (handle.respond === undefined) { + throw new Error(`Backend ${backend} doesn't run calls that wait for an answer.`); + } + await handle.respond({ id, callId, value }); + } + 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 050873fc1..32be1a276 100644 --- a/packages/computer/src/runtime/types.ts +++ b/packages/computer/src/runtime/types.ts @@ -28,6 +28,27 @@ export interface WorkspaceModuleCallContext { * the same id, so a function can use it as an idempotency key. */ readonly callId: string; + /** + * Pause the run until a person or another system answers, and return + * the answer. The run is released while it waits, survives a deploy, + * and resumes once `workspace.runtime.respond()` answers this call: + * it replays to this call, and this function runs again with the same + * `callId`, this time getting the answer straight away. Anything the + * function does before `interrupt()` therefore runs twice, so keep it + * safe to repeat. Only a run with `onInterrupt: "replay"` can wait; + * anywhere else this throws. + * + * ```ts + * postComment: async ([issue, body], ctx) => { + * const { approved } = await ctx.interrupt({ kind: "approval", issue, body }); + * if (!approved) throw new Error("The comment was rejected."); + * return github.comment(issue, body, { idempotencyKey: ctx.callId }); + * } + * ``` + * + * @param payload What the answer is for, as `pending()` lists it. + */ + interrupt(payload: WorkspaceRuntimeValue): Promise; } /** @@ -283,6 +304,19 @@ export interface ModuleExecutionInput { */ export type WorkspaceRuntimeInterruptPolicy = "fail" | "restart" | "replay"; +/** A host module call waiting for an answer, from `ctx.interrupt()`. */ +export interface WorkspaceRuntimePendingCall { + /** The execution the call belongs to. */ + readonly id: string; + readonly backend: string; + /** Pass this to `respond()` to answer the call. */ + readonly callId: string; + /** What the call passed to `ctx.interrupt()`. */ + readonly payload: WorkspaceRuntimeValue; + /** When the run started waiting, in milliseconds since the epoch. */ + readonly since: number; +} + export interface ModuleExecutionEnvelope { id: string; // Identity of the process-local backend runtime that owns this @@ -313,6 +347,9 @@ export interface WorkspaceModuleBackendHandle { // Settle the runs a restart interrupted. A backend that recovers // when it connects needs nothing more here. recover?(): Promise; + // Calls waiting for an answer, and answering one. + pending?(): Promise[]>; + respond?(input: { id: string; callId: string; value: WorkspaceRuntimeValue }): Promise; // Tear down a backend-owned transport. The command adapter omits // it: a command backend's transport is closed through its // BackendHandle, not through the adapter the runtime consumes. diff --git a/packages/computer/src/tools/ai-sdk/bindings.test.ts b/packages/computer/src/tools/ai-sdk/bindings.test.ts index 233fe7c69..5fd1e0fd4 100644 --- a/packages/computer/src/tools/ai-sdk/bindings.test.ts +++ b/packages/computer/src/tools/ai-sdk/bindings.test.ts @@ -15,6 +15,7 @@ function callContext(): WorkspaceModuleCallContext { deadline: Date.now() + 60_000, access: "read-write", callId: "test-call", + 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 3894c1d33..93f679a65 100644 --- a/packages/computer/src/tools/pi-ai/bindings.test.ts +++ b/packages/computer/src/tools/pi-ai/bindings.test.ts @@ -18,6 +18,7 @@ function callContext( deadline: Date.now() + 60_000, access, callId: "test-call", + 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 7cc3dc370..a65ca2b6f 100644 --- a/packages/computer/tests/script-runner-worker.ts +++ b/packages/computer/tests/script-runner-worker.ts @@ -149,6 +149,27 @@ function javascriptBackend( await new Promise((resolve) => setTimeout(resolve, Number(args[0]))); return countCall(String(args[1] ?? "wait")); }, + // Counts when it starts and again when it settles, and stops early + // if its call is aborted. + async slowTick(args, ctx) { + countCall(`start:${String(args[1])}`); + await new Promise((resolve, reject) => { + const timer = setTimeout(resolve, Number(args[0])); + ctx.signal.addEventListener("abort", () => { + clearTimeout(timer); + reject(ctx.signal.reason); + }); + }); + return countCall(String(args[1])); + }, + // Waits for an answer, then returns it. + async approve(args, ctx) { + const action = String(args[0]); + countCall(`ask:${action}`); + const answer = await ctx.interrupt({ kind: "approval", action }); + countCall(`answered:${action}`); + return answer; + }, async marker() { return { __workspace_codec__: { version: 1, type: "bytes", data: [1] }, @@ -281,6 +302,36 @@ export class HostDO extends DurableObject { return handle.id; } + async startWaiting(input: { + source: string; + id: string; + onInterrupt?: "fail" | "restart" | "replay"; + timeoutMs?: number; + }) { + durableCalls.clear(); + const handle = await this.#workspace.runtime.exec(input.source, { + backend: "worker-javascript", + id: input.id, + onInterrupt: input.onInterrupt ?? "replay", + timeoutMs: input.timeoutMs, + }); + void handle.result().catch(() => undefined); + return handle.id; + } + + pending() { + return this.#workspace.runtime.pending(); + } + + async respond(id: string, callId: string, value: WorkspaceRuntimeValue) { + try { + await this.#workspace.runtime.respond(id, callId, value); + return { ok: true as const }; + } catch (error) { + return { ok: false as const, code: (error as { code?: string }).code }; + } + } + // Restart the object, as a deploy or crash would. restart(open?: string) { gateOpen = true; @@ -544,6 +595,32 @@ export default class extends WorkerEntrypoint { return new Response(null, { status: 204 }); } + if (url.pathname === "/interrupt-start") { + return Response.json({ + id: await stub.startWaiting( + (await request.json()) as { + source: string; + id: string; + onInterrupt?: "fail" | "restart" | "replay"; + timeoutMs?: number; + }, + ), + }); + } + + if (url.pathname === "/interrupt-pending") { + return Response.json(await stub.pending()); + } + + if (url.pathname === "/interrupt-respond") { + const body = (await request.json()) as { + id: string; + callId: string; + value: WorkspaceRuntimeValue; + }; + return Response.json(await stub.respond(body.id, body.callId, body.value)); + } + if (url.pathname === "/durable-calls") { return Response.json(await stub.durableCalls()); } diff --git a/packages/computer/tests/script-runner.test.ts b/packages/computer/tests/script-runner.test.ts index 56dc8bb5b..39c29ee03 100644 --- a/packages/computer/tests/script-runner.test.ts +++ b/packages/computer/tests/script-runner.test.ts @@ -330,6 +330,7 @@ describe("WorkspaceRuntime", () => { const text = await response.text(); expect(response.status, text).toBe(200); expect(JSON.parse(text).result.value).toEqual([ + "approve", "delete", "echo", "gate", @@ -338,6 +339,7 @@ describe("WorkspaceRuntime", () => { "largeError", "marker", "slow", + "slowTick", "sum", "tick", "unsafeGate", @@ -1341,3 +1343,180 @@ describe("durable executions", () => { expect(await result("durable-recover")).toMatchObject({ status: "completed" }); }); }); + +describe("interrupts", () => { + type Pending = { id: string; backend: string; callId: string; payload: unknown; since: number }; + + async function start(id: string, source: string, options: Record = {}) { + const response = await SELF.fetch("https://example.test/interrupt-start", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ id, source, ...options }), + }); + expect(response.status, await response.clone().text()).toBe(200); + } + + async function pending(id: string): Promise { + const response = await SELF.fetch("https://example.test/interrupt-pending"); + return ((await response.json()) as Pending[]).filter((call) => call.id === id); + } + + // The first of a run's calls to wait, other than `skip`, once one does. + async function waitingCall(id: string, skip?: string): Promise { + for (let attempt = 0; attempt < 200; attempt += 1) { + const call = (await pending(id)).find((item) => item.callId !== skip); + if (call !== undefined) return call; + await new Promise((resolve) => setTimeout(resolve, 20)); + } + throw new Error(`no call of ${id} is waiting`); + } + + async function respond(id: string, callId: string, value: unknown) { + const response = await SELF.fetch("https://example.test/interrupt-respond", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ id, callId, value }), + }); + return (await response.json()) as { ok: boolean; code?: string }; + } + + async function result(id: string) { + const response = await SELF.fetch( + `https://example.test/runtime-get?id=${encodeURIComponent(id)}`, + ); + return (await response.json()) as { + status: string; + stdout: string; + stderr: string; + value?: unknown; + }; + } + + async function calls() { + return (await (await SELF.fetch("https://example.test/durable-calls")).json()) as Record< + string, + number + >; + } + + const approval = ` + import { approve, tick } from "ws:test-host"; + export default async function () { + console.log("before", await tick("a")); + const answer = await approve("deploy"); + console.log("after", answer.ok); + return answer; + } + `; + + it("waits for an answer, then resumes where it was", async () => { + await start("wait-answer", approval); + const call = await waitingCall("wait-answer"); + expect(call).toMatchObject({ + backend: "worker-javascript", + callId: expect.stringMatching(/^wait-answer:\d+$/), + payload: { kind: "approval", action: "deploy" }, + }); + await expect(respond("wait-answer", call.callId, { ok: true })).resolves.toEqual({ ok: true }); + expect(await result("wait-answer")).toMatchObject({ + status: "completed", + stdout: "before 1\nafter true\n", + value: { ok: true }, + }); + // The function ran again to get its answer; nothing else repeated. + expect(await calls()).toEqual({ a: 1, "ask:deploy": 2, "answered:deploy": 1 }); + expect(await pending("wait-answer")).toEqual([]); + await expect(respond("wait-answer", call.callId, { ok: true })).resolves.toEqual({ + ok: false, + code: "ENOTWAITING", + }); + }); + + it("keeps waiting across a restart", async () => { + await start("wait-restart", approval); + await waitingCall("wait-restart"); + await SELF.fetch("https://example.test/durable-restart", { method: "POST" }); + const call = await waitingCall("wait-restart"); + expect(call.payload).toEqual({ kind: "approval", action: "deploy" }); + await respond("wait-restart", call.callId, { ok: false }); + expect(await result("wait-restart")).toMatchObject({ + status: "completed", + stdout: "before 1\nafter false\n", + }); + expect(await calls()).toEqual({ a: 1, "ask:deploy": 2, "answered:deploy": 1 }); + }); + + it("asks calls made together one after another", async () => { + await start( + "wait-both", + ` + import { approve } from "ws:test-host"; + export default async () => (await Promise.all([approve("one"), approve("two")])).map((a) => a.v); + `, + ); + // The second call reaches the host once the run is being released, + // so it waits until the first is answered and the run resumes. + const first = await waitingCall("wait-both"); + expect(first.payload).toEqual({ kind: "approval", action: "one" }); + await respond("wait-both", first.callId, { v: "first" }); + const second = await waitingCall("wait-both", first.callId); + expect(second.payload).toEqual({ kind: "approval", action: "two" }); + await respond("wait-both", second.callId, { v: "second" }); + expect(await result("wait-both")).toMatchObject({ + status: "completed", + value: ["first", "second"], + }); + expect(await calls()).toEqual({ + "ask:one": 2, + "answered:one": 1, + "ask:two": 2, + "answered:two": 1, + }); + }); + + it("lets other calls in flight finish before it releases the run", async () => { + await start( + "wait-inflight", + ` + import { approve, slowTick } from "ws:test-host"; + // slowTick reaches the host first, so it's in flight when approve interrupts. + export default async () => Promise.all([slowTick(300, "slow"), approve("x")]); + `, + ); + const call = await waitingCall("wait-inflight"); + await respond("wait-inflight", call.callId, { ok: true }); + expect(await result("wait-inflight")).toMatchObject({ + status: "completed", + value: [1, { ok: true }], + }); + // Journaled when it finished, so the resumed run didn't start it again. + expect(await calls()).toMatchObject({ "start:slow": 1, slow: 1 }); + }); + + it("doesn't count waiting toward the run's timeout", async () => { + await start("wait-timeout", approval, { timeoutMs: 1500 }); + const call = await waitingCall("wait-timeout"); + await new Promise((resolve) => setTimeout(resolve, 2000)); + await respond("wait-timeout", call.callId, { ok: true }); + expect(await result("wait-timeout")).toMatchObject({ status: "completed" }); + }); + + it("can be killed while it waits", async () => { + await start("wait-kill", approval); + const call = await waitingCall("wait-kill"); + await SELF.fetch("https://example.test/runtime-kill?id=wait-kill", { method: "POST" }); + expect(await result("wait-kill")).toMatchObject({ status: "cancelled" }); + expect(await pending("wait-kill")).toEqual([]); + await expect(respond("wait-kill", call.callId, { ok: true })).resolves.toMatchObject({ + code: "ENOTWAITING", + }); + }); + + it("throws in a run that can't be released", async () => { + await start("wait-fail", approval, { onInterrupt: "fail" }); + const settled = await result("wait-fail"); + expect(settled.status).toBe("failed"); + // The test backend keeps 64 bytes of standard error. + expect(settled.stderr).toContain("ctx.interrupt() needs a replayable run"); + }); +}); From 3e50a4b41c97d6f1b56cde676bbb17a98acd5169 Mon Sep 17 00:00:00 2001 From: scuffi Date: Mon, 19 Oct 2026 15:16:42 +0100 Subject: [PATCH 02/12] docs: Describe waiting for an answer in a replayable run --- docs/17_isolate_javascript.md | 31 +++++++++++++++++++++++++++++++ packages/computer/README.md | 2 +- 2 files changed, 32 insertions(+), 1 deletion(-) diff --git a/docs/17_isolate_javascript.md b/docs/17_isolate_javascript.md index 7cb0d1b3c..7889c62a8 100644 --- a/docs/17_isolate_javascript.md +++ b/docs/17_isolate_javascript.md @@ -195,6 +195,37 @@ modules: { } ``` +### Waiting for an answer + +A host module function can pause a replayable run until someone answers, for example to approve an action. The code being run just calls the function and gets its result, however long the answer takes: + +```ts +modules: { + "ws:github": { + postComment: async ([issue, body], ctx) => { + const { approved } = await ctx.interrupt({ kind: "approval", issue, body }); + if (!approved) throw new Error("The comment was rejected."); + return github.comment(issue, body, { idempotencyKey: ctx.callId }); + }, + }, +} +``` + +`ctx.interrupt(payload)` releases the run's isolate once the run's other calls in flight finish. The run stays running, and time spent waiting doesn't count toward its timeout. `workspace.runtime.pending()` lists the calls waiting, with their payloads, and `respond()` answers one: + +```ts +for (const call of await workspace.runtime.pending()) { + await workspace.runtime.respond(call.id, call.callId, { approved: true }); +} +``` + +Once none of its calls are waiting, the run replays to the call and the function runs again with the same `callId`. This time `interrupt()` returns the answer straight away, and the function carries on. Anything it does before `interrupt()` runs twice, so keep that part safe to repeat. A waiting run survives a deploy like any interrupted run, and `killExec()` ends one. + +- Only a run with `onInterrupt: "replay"` can wait. Anywhere else, `interrupt()` throws, so a Workspace with modules that interrupt should make `"replay"` its backend's default. +- Calls that interrupt together are answered one after another. A run is released with the calls that were in flight, and a call it makes after that waits once it resumes. +- `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. + ### Limits - A run gets three attempts, and has to finish within its original timeout, counted from when it first started. diff --git a/packages/computer/README.md b/packages/computer/README.md index fb518a2e8..3721cb503 100644 --- a/packages/computer/README.md +++ b/packages/computer/README.md @@ -31,7 +31,7 @@ removed in a future release; each one names its replacement. | Observability hook and its Cloudflare adapter | `@cloudflare/computer`, `@cloudflare/computer/observe/cloudflare` | Stable | | `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()`, and `unsafeToRepeat()` | `@cloudflare/computer`, `@cloudflare/computer/backends/worker-javascript` | Experimental | +| 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/worker-javascript` | Experimental | | `WorkerShellBackend` and the shell command groups | `@cloudflare/computer/backends/worker-shell`, `@cloudflare/computer/shell/*` | Experimental | | The `ws:container` host module | `@cloudflare/computer/modules/container` | Experimental | | `workspace.push()`, `workspace.pull()`, and the sync protocol between the durable object and `computerd` | `@cloudflare/computer` | Experimental | From 313e69a834e3dd030de9f66a820f048cb5ea2333 Mon Sep 17 00:00:00 2001 From: scuffi Date: Mon, 19 Oct 2026 16:54:31 +0100 Subject: [PATCH 03/12] computer: Let a host module call interrupt only once The journal keeps one answer per call. A function that called ctx.interrupt() twice joined its first wait the second time, losing the second payload, and on resume both calls returned the first answer, so one approval could stand for two. A second interrupt() in the same call now throws. --- docs/17_isolate_javascript.md | 1 + packages/computer/src/runtime/bridge.test.ts | 18 ++++++++++++++++-- packages/computer/src/runtime/bridge.ts | 12 +++++++++++- packages/computer/src/runtime/types.ts | 8 ++++++-- 4 files changed, 34 insertions(+), 5 deletions(-) diff --git a/docs/17_isolate_javascript.md b/docs/17_isolate_javascript.md index 7889c62a8..1a1e599aa 100644 --- a/docs/17_isolate_javascript.md +++ b/docs/17_isolate_javascript.md @@ -221,6 +221,7 @@ for (const call of await workspace.runtime.pending()) { Once none of its calls are waiting, the run replays to the call and the function runs again with the same `callId`. This time `interrupt()` returns the answer straight away, and the function carries on. Anything it does before `interrupt()` runs twice, so keep that part safe to repeat. A waiting run survives a deploy like any interrupted run, and `killExec()` ends one. +- A call can interrupt once. A second `interrupt()` in the same call throws. - Only a run with `onInterrupt: "replay"` can wait. Anywhere else, `interrupt()` throws, so a Workspace with modules that interrupt should make `"replay"` its backend's default. - Calls that interrupt together are answered one after another. A run is released with the calls that were in flight, and a call it makes after that waits once it resumes. - `respond()` throws `ENOTWAITING` for a call that isn't waiting, such as one already answered, so answering twice changes nothing. diff --git a/packages/computer/src/runtime/bridge.test.ts b/packages/computer/src/runtime/bridge.test.ts index b6b4f787a..91c17ead2 100644 --- a/packages/computer/src/runtime/bridge.test.ts +++ b/packages/computer/src/runtime/bridge.test.ts @@ -249,7 +249,7 @@ describe("WorkspaceRuntimeBridge assertResult", () => { }); describe("WorkspaceRuntimeBridge journal", () => { - function journaled(steps: ReturnType[]) { + function journaled(steps: ReturnType[], maxTotalResponseBytes = 4) { const commits: number[] = []; const responses: unknown[] = []; const journal: BridgeJournal = { @@ -271,7 +271,7 @@ describe("WorkspaceRuntimeBridge journal", () => { const logs: unknown[] = []; const contexts: WorkspaceModuleCallContext[] = []; const target = new WorkspaceRuntimeBridge({} as WorkspaceRuntimeCapability, { - maxTotalResponseBytes: 4, + maxTotalResponseBytes, hostModules: new Map([ [ "ws:test", @@ -281,6 +281,10 @@ describe("WorkspaceRuntimeBridge journal", () => { return "ok"; }, charge: unsafeToRepeat(async () => "c"), + twice: async (_args: unknown, context: WorkspaceModuleCallContext) => [ + await context.interrupt("first"), + await context.interrupt("second"), + ], }, ], ]), @@ -362,6 +366,16 @@ describe("WorkspaceRuntimeBridge journal", () => { expect(diverged).toHaveLength(1); }); + it("lets a call interrupt only once", async () => { + const { target } = journaled( + [{ kind: "live", seq: 1, rerun: false, answer: { value: "yes" } }], + 4096, + ); + await expect(message(target.call("host/ws:test.twice", args))).resolves.toContain( + "ctx.interrupt() can only be called once per host module call.", + ); + }); + it("passes a stalled replay to the journal", () => { const { target, diverged } = journaled([]); target.reportStall(); diff --git a/packages/computer/src/runtime/bridge.ts b/packages/computer/src/runtime/bridge.ts index 051d6b4c0..84822726f 100644 --- a/packages/computer/src/runtime/bridge.ts +++ b/packages/computer/src/runtime/bridge.ts @@ -194,13 +194,23 @@ export class WorkspaceRuntimeBridge extends RpcTarget { const abort = new AbortController(); this.#abortControllers.add(abort); const deadline = Date.now() + this.#maxCallDurationMs; + // The journal holds one answer per call, so a call can wait only once. + let interrupted = false; const operation = respond(async () => { if (name.startsWith("host/")) { return this.#callHostModule(name, args, { signal: abort.signal, deadline, callId, - interrupt: (payload) => this.#interrupt(step, abort, payload), + interrupt: (payload) => { + if (interrupted) { + return Promise.reject( + new Error("ctx.interrupt() can only be called once per host module call."), + ); + } + interrupted = true; + return this.#interrupt(step, abort, payload); + }, access: this.#capability.access, resolvePath: (path, options) => this.#capability.resolveConfined(path, options?.allowMissing ?? false), diff --git a/packages/computer/src/runtime/types.ts b/packages/computer/src/runtime/types.ts index 32be1a276..274e5e473 100644 --- a/packages/computer/src/runtime/types.ts +++ b/packages/computer/src/runtime/types.ts @@ -36,7 +36,8 @@ export interface WorkspaceModuleCallContext { * `callId`, this time getting the answer straight away. Anything the * function does before `interrupt()` therefore runs twice, so keep it * safe to repeat. Only a run with `onInterrupt: "replay"` can wait; - * anywhere else this throws. + * anywhere else this throws. A call can interrupt once, and a second + * `interrupt()` throws. * * ```ts * postComment: async ([issue, body], ctx) => { @@ -304,7 +305,10 @@ export interface ModuleExecutionInput { */ export type WorkspaceRuntimeInterruptPolicy = "fail" | "restart" | "replay"; -/** A host module call waiting for an answer, from `ctx.interrupt()`. */ +/** + * A host module call waiting for an answer, from `ctx.interrupt()`. + * Experimental: it may change in any release. + */ export interface WorkspaceRuntimePendingCall { /** The execution the call belongs to. */ readonly id: string; From 4f9cf7d288fa284e03acd810d71f755a6de206d0 Mon Sep 17 00:00:00 2001 From: scuffi Date: Mon, 19 Oct 2026 16:54:31 +0100 Subject: [PATCH 04/12] computer: Test an answer that arrives before the run is released The run resumes only once its isolate is released, so an answer that lands while the run's other calls are still finishing has to wait for the release to end. Nothing checked that it does, and if it didn't the run would stay running without ever resuming. --- packages/computer/tests/script-runner.test.ts | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/packages/computer/tests/script-runner.test.ts b/packages/computer/tests/script-runner.test.ts index 39c29ee03..635c2f37d 100644 --- a/packages/computer/tests/script-runner.test.ts +++ b/packages/computer/tests/script-runner.test.ts @@ -1493,6 +1493,24 @@ describe("interrupts", () => { expect(await calls()).toMatchObject({ "start:slow": 1, slow: 1 }); }); + it("resumes when the answer arrives before the run is released", async () => { + await start( + "wait-early", + ` + import { approve, slowTick } from "ws:test-host"; + export default async () => Promise.all([slowTick(1000, "early"), approve("x")]); + `, + ); + const call = await waitingCall("wait-early"); + await respond("wait-early", call.callId, { ok: true }); + // slowTick is still running, so the run hadn't been released yet. + expect(await calls()).not.toHaveProperty("early"); + expect(await result("wait-early")).toMatchObject({ + status: "completed", + value: [1, { ok: true }], + }); + }); + it("doesn't count waiting toward the run's timeout", async () => { await start("wait-timeout", approval, { timeoutMs: 1500 }); const call = await waitingCall("wait-timeout"); From b1ec24be761a94904d3849233bfb2bfa5e0f57b8 Mon Sep 17 00:00:00 2001 From: scuffi Date: Mon, 19 Oct 2026 16:54:31 +0100 Subject: [PATCH 05/12] computer, docs: Mark waiting for an answer experimental ctx.interrupt(), runtime.pending(), runtime.respond() and the pending call type now say in their JSDoc that they may change in any release, as restart and replay already do, and the docs section says so too. --- docs/17_isolate_javascript.md | 2 +- packages/computer/src/runtime/runtime.ts | 3 ++- packages/computer/src/runtime/types.ts | 2 +- 3 files changed, 4 insertions(+), 3 deletions(-) diff --git a/docs/17_isolate_javascript.md b/docs/17_isolate_javascript.md index 1a1e599aa..63754430e 100644 --- a/docs/17_isolate_javascript.md +++ b/docs/17_isolate_javascript.md @@ -197,7 +197,7 @@ modules: { ### Waiting for an answer -A host module function can pause a replayable run until someone answers, for example to approve an action. The code being run just calls the function and gets its result, however long the answer takes: +A host module function can pause a replayable run until someone answers, for example to approve an action. Like replay, this is experimental. The code being run just calls the function and gets its result, however long the answer takes: ```ts modules: { diff --git a/packages/computer/src/runtime/runtime.ts b/packages/computer/src/runtime/runtime.ts index ecd5c93ec..bc6d7fd25 100644 --- a/packages/computer/src/runtime/runtime.ts +++ b/packages/computer/src/runtime/runtime.ts @@ -177,6 +177,7 @@ export class WorkspaceRuntime { /** * Host module calls waiting for an answer, from `ctx.interrupt()`, * across every backend that supports it, oldest wait first. + * Experimental: it may change in any release. */ async pending(): Promise { const calls: WorkspaceRuntimePendingCall[] = []; @@ -193,7 +194,7 @@ export class WorkspaceRuntime { * none of its calls are still waiting, and the call gets `value` as * what `interrupt()` returns. Throws `ENOTWAITING` if the call isn't * waiting, for example because it was already answered or the run - * was killed. + * was killed. Experimental: it may change in any release. */ async respond( id: string, diff --git a/packages/computer/src/runtime/types.ts b/packages/computer/src/runtime/types.ts index 274e5e473..18c7fbc19 100644 --- a/packages/computer/src/runtime/types.ts +++ b/packages/computer/src/runtime/types.ts @@ -37,7 +37,7 @@ export interface WorkspaceModuleCallContext { * function does before `interrupt()` therefore runs twice, so keep it * safe to repeat. Only a run with `onInterrupt: "replay"` can wait; * anywhere else this throws. A call can interrupt once, and a second - * `interrupt()` throws. + * `interrupt()` throws. Experimental: it may change in any release. * * ```ts * postComment: async ([issue, body], ctx) => { From 4f7338de6e5a41a0064b978e7978b664d4ab6e72 Mon Sep 17 00:00:00 2001 From: scuffi Date: Tue, 20 Oct 2026 11:15:22 +0100 Subject: [PATCH 06/12] computer: Tell the backend's onWait when a call starts waiting An app had to poll runtime.pending() to learn that a run was waiting, and had to know which object to ask. The JavaScript backend now takes an onWait function, called with the fields pending() lists once a wait is saved, so the app can message whoever answers or index the wait itself. It's called at most once per wait. A crash between saving the wait and calling it loses the call, so pending() stays the source of truth. It runs on its own rather than inside the interrupted call, so a slow or failing hook can't hold up the run's release, and what it throws is logged. --- docs/17_isolate_javascript.md | 2 ++ .../src/backends/worker-javascript/journal.ts | 9 +++-- .../worker-javascript/worker-javascript.ts | 33 ++++++++++++++++++- packages/computer/src/runtime/bridge.test.ts | 2 +- packages/computer/src/runtime/bridge.ts | 11 ++++--- .../computer/tests/script-runner-worker.ts | 18 +++++++++- packages/computer/tests/script-runner.test.ts | 11 +++++++ 7 files changed, 75 insertions(+), 11 deletions(-) diff --git a/docs/17_isolate_javascript.md b/docs/17_isolate_javascript.md index 79600472e..3dc77e384 100644 --- a/docs/17_isolate_javascript.md +++ b/docs/17_isolate_javascript.md @@ -219,6 +219,8 @@ for (const call of await workspace.runtime.pending()) { } ``` +To hear about a wait when it starts, for example to message whoever answers, give the backend an `onWait` function. It gets the same fields, at most once per wait, so `pending()` stays the source of truth. + Once none of its calls are waiting, the run replays to the call and the function runs again with the same `callId`. This time `interrupt()` returns the answer straight away, and the function carries on. Anything it does before `interrupt()` runs twice, so keep that part safe to repeat. A waiting run survives a deploy like any interrupted run, and `killExec()` ends one. - A call can interrupt once. A second `interrupt()` in the same call throws. diff --git a/packages/computer/src/backends/worker-javascript/journal.ts b/packages/computer/src/backends/worker-javascript/journal.ts index 896aad09f..f4b8b39fc 100644 --- a/packages/computer/src/backends/worker-javascript/journal.ts +++ b/packages/computer/src/backends/worker-javascript/journal.ts @@ -611,7 +611,8 @@ export class ExecutionJournal implements BridgeJournal { * committed: the run is released, and the call runs again once * answered. */ - interrupt(seq: number, payload: unknown): void { + /** Mark a call waiting, and return when its run started waiting. */ + interrupt(seq: number, payload: unknown): number { this.#db.run( `UPDATE workspace_runtime_journal SET state = 'waiting', payload = ? WHERE backend = ? AND execution_id = ? AND seq = ?`, @@ -620,13 +621,15 @@ export class ExecutionJournal implements BridgeJournal { this.#id, seq, ); - this.#db.run( + const [row] = this.#db.all<{ waiting_since: number }>( `UPDATE workspace_runtime_durable SET waiting_since = COALESCE(waiting_since, ?) - WHERE backend = ? AND execution_id = ?`, + WHERE backend = ? AND execution_id = ? + RETURNING waiting_since`, Date.now(), this.#backend, this.#id, ); + return Number(row?.waiting_since ?? Date.now()); } diverged(message: string): void { diff --git a/packages/computer/src/backends/worker-javascript/worker-javascript.ts b/packages/computer/src/backends/worker-javascript/worker-javascript.ts index 312f82d25..e1d023eab 100644 --- a/packages/computer/src/backends/worker-javascript/worker-javascript.ts +++ b/packages/computer/src/backends/worker-javascript/worker-javascript.ts @@ -118,6 +118,16 @@ export interface WorkerJavaScriptBackendOptions { * release. */ onInterrupt?: WorkspaceRuntimeInterruptPolicy; + /** + * Called when a host module call starts waiting in `ctx.interrupt()`, + * with the fields `runtime.pending()` lists, for example to ask someone + * for the answer. It's called once the wait is saved, and at most once + * per wait: a crash in between loses the call. Treat `pending()` as the + * source of truth, for example by checking it when the object starts. + * What it throws is logged and otherwise ignored. Experimental: it may + * change in any release. + */ + onWait?: (call: WorkspaceRuntimePendingCall) => 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 @@ -588,6 +598,24 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { this.#resumeIfAnswered(record); } + #notifyWait(record: ExecutionRecord, call: { seq: number; payload: unknown; since: number }) { + const { onWait } = this.#options; + if (onWait === undefined) return; + const pending: WorkspaceRuntimePendingCall = { + id: record.id, + backend: this.#backendId, + callId: `${record.id}:${call.seq}`, + payload: call.payload as WorkspaceRuntimeValue, + since: call.since, + }; + // After the current call, so the hook can't hold up the run's release. + void Promise.resolve() + .then(() => onWait(pending)) + .catch((error: unknown) => { + console.error({ message: "onWait failed", id: record.id, error: String(error) }); + }); + } + #resumeIfAnswered(record: ExecutionRecord) { if (record.status !== "running" || record.waiting !== "released") return; const saved = this.#loadDurable(record); @@ -913,7 +941,10 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { maxResultBytes: this.#options.maxResultBytes, onAttachOutput: (readable) => this.#pumpFrames(record, readable), callIdPrefix: record.id, - onWait: () => void this.#release(record), + onWait: (call) => { + this.#notifyWait(record, call); + void this.#release(record); + }, ...hostEgressFetch(this.#options.egress), ...(spec.journal === undefined ? {} : { journal: spec.journal }), }); diff --git a/packages/computer/src/runtime/bridge.test.ts b/packages/computer/src/runtime/bridge.test.ts index 91c17ead2..a73405583 100644 --- a/packages/computer/src/runtime/bridge.test.ts +++ b/packages/computer/src/runtime/bridge.test.ts @@ -263,7 +263,7 @@ describe("WorkspaceRuntimeBridge journal", () => { if (!Array.isArray(events)) throw new Error("malformed event log"); logs.push([position, events]); }, - interrupt: () => undefined, + interrupt: () => 0, diverged: (message) => diverged.push(message), stalled: () => diverged.push("stalled"), }; diff --git a/packages/computer/src/runtime/bridge.ts b/packages/computer/src/runtime/bridge.ts index 84822726f..0e8b8a375 100644 --- a/packages/computer/src/runtime/bridge.ts +++ b/packages/computer/src/runtime/bridge.ts @@ -32,7 +32,7 @@ export class WorkspaceRuntimeBridge extends RpcTarget { readonly #responses = new Map, AbortController>(); // Calls that interrupted, by their controller. readonly #waiting = new Set(); - readonly #onWait?: () => void; + readonly #onWait?: (call: { seq: number; payload: unknown; since: number }) => void; // Set once a call interrupts: new calls are held, never started. #holding = false; #cancelled = false; @@ -60,7 +60,7 @@ export class WorkspaceRuntimeBridge extends RpcTarget { // Prefix of every call's callId, such as the execution id. callIdPrefix?: string; // Called when a call interrupts, to release the run. - onWait?: () => void; + onWait?: (call: { seq: number; payload: unknown; since: number }) => void; } = {}, ) { super(); @@ -349,13 +349,14 @@ export class WorkspaceRuntimeBridge extends RpcTarget { if (this.#cancelled) return Promise.reject(new Error("Workspace execution is being cancelled.")); if (!this.#waiting.has(abort)) { + let since: number; try { - this.#journal.interrupt(step.seq, payload); + since = this.#journal.interrupt(step.seq, payload); } catch (error) { return Promise.reject(error); } this.#waiting.add(abort); - this.#onWait?.(); + this.#onWait?.({ seq: step.seq, payload, since }); } // Settles only when the run is released and the call aborted. return new Promise((_, reject) => { @@ -621,7 +622,7 @@ export interface BridgeJournal { /** Record a live call's response, and return its time and ordinal. */ commit(seq: number, response: BridgeResponse): { at: number; ordinal: number }; /** Mark a live call as waiting for an answer. */ - interrupt(seq: number, payload: unknown): void; + interrupt(seq: number, payload: unknown): number; /** Save a chunk of the isolate's event log. Throws on a malformed one. */ record(position: unknown, events: unknown): void; /** Called when a call doesn't match the journal. */ diff --git a/packages/computer/tests/script-runner-worker.ts b/packages/computer/tests/script-runner-worker.ts index a65ca2b6f..b75767190 100644 --- a/packages/computer/tests/script-runner-worker.ts +++ b/packages/computer/tests/script-runner-worker.ts @@ -6,7 +6,11 @@ import { type WorkerJavaScriptBackendOptions, } from "../src/backends/worker-javascript/index.js"; import { createGitClient } from "../src/git/index.js"; -import type { WorkspaceRuntimeValue, WorkspaceStub } from "../src/index.js"; +import type { + WorkspaceRuntimePendingCall, + WorkspaceRuntimeValue, + WorkspaceStub, +} from "../src/index.js"; import { unsafeToRepeat, Workspace } from "../src/index.js"; import { createArtifactsModule } from "../src/modules/artifacts.js"; import { createContainerModule } from "../src/modules/container.js"; @@ -61,6 +65,8 @@ function fakeContainerBackend(): WorkspaceBackend { // state outlives the object, so it counts host calls across the restart // and holds the gate a run waits on until the object restarts. const durableCalls = new Map(); +// Each call onWait was told about. +const waits: WorkspaceRuntimePendingCall[] = []; let gateOpen = false; // Named gates a run can wait on after the first, for a second restart. const opened = new Set(); @@ -90,6 +96,7 @@ function javascriptBackend( ...overrides, maxStdioBytes: overrides.maxStdioBytes ?? 64, maxCapabilityBytes: 1024, + onWait: (call) => void waits.push(call), maxConcurrentCapabilityCalls: 2, modules: { "math-kit": "export const double = (value) => value * 2;", @@ -309,6 +316,7 @@ export class HostDO extends DurableObject { timeoutMs?: number; }) { durableCalls.clear(); + waits.length = 0; const handle = await this.#workspace.runtime.exec(input.source, { backend: "worker-javascript", id: input.id, @@ -323,6 +331,10 @@ export class HostDO extends DurableObject { return this.#workspace.runtime.pending(); } + waits() { + return waits; + } + async respond(id: string, callId: string, value: WorkspaceRuntimeValue) { try { await this.#workspace.runtime.respond(id, callId, value); @@ -608,6 +620,10 @@ export default class extends WorkerEntrypoint { }); } + if (url.pathname === "/interrupt-waits") { + return Response.json(await stub.waits()); + } + 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 635c2f37d..7973cceae 100644 --- a/packages/computer/tests/script-runner.test.ts +++ b/packages/computer/tests/script-runner.test.ts @@ -1519,6 +1519,17 @@ describe("interrupts", () => { expect(await result("wait-timeout")).toMatchObject({ status: "completed" }); }); + it("tells onWait about a wait once, even across a restart and resume", async () => { + await start("wait-notify", approval); + const call = await waitingCall("wait-notify"); + await SELF.fetch("https://example.test/durable-restart", { method: "POST" }); + await waitingCall("wait-notify"); + await respond("wait-notify", call.callId, { ok: true }); + expect(await result("wait-notify")).toMatchObject({ status: "completed" }); + const waits = await (await SELF.fetch("https://example.test/interrupt-waits")).json(); + expect(waits).toEqual([call]); + }); + it("can be killed while it waits", async () => { await start("wait-kill", approval); const call = await waitingCall("wait-kill"); From bab8d02c574259e7b837fbbea144e37f09046c7b Mon Sep 17 00:00:00 2001 From: scuffi Date: Tue, 20 Oct 2026 11:25:23 +0100 Subject: [PATCH 07/12] computer: Add pending() and respond() to the runtime stub Approval UIs usually live in a Worker that calls the Durable Object, so without these every app wrote its own RPC methods to list and answer waiting calls. Both pass plain data, so the stub hands out nothing that needs disposing. --- docs/17_isolate_javascript.md | 2 +- packages/computer/src/stub.test.ts | 13 ++++++++++++- packages/computer/src/stub.ts | 16 ++++++++++++++++ 3 files changed, 29 insertions(+), 2 deletions(-) diff --git a/docs/17_isolate_javascript.md b/docs/17_isolate_javascript.md index 3dc77e384..d3b0c671d 100644 --- a/docs/17_isolate_javascript.md +++ b/docs/17_isolate_javascript.md @@ -219,7 +219,7 @@ for (const call of await workspace.runtime.pending()) { } ``` -To hear about a wait when it starts, for example to message whoever answers, give the backend an `onWait` function. It gets the same fields, at most once per wait, so `pending()` stays the source of truth. +To hear about a wait when it starts, for example to message whoever answers, give the backend an `onWait` function. It gets the same fields, at most once per wait, so `pending()` stays the source of truth. A Worker that calls the object can reach `pending()` and `respond()` through `workspace.stub().runtime`. Once none of its calls are waiting, the run replays to the call and the function runs again with the same `callId`. This time `interrupt()` returns the answer straight away, and the function carries on. Anything it does before `interrupt()` runs twice, so keep that part safe to repeat. A waiting run survives a deploy like any interrupted run, and `killExec()` ends one. diff --git a/packages/computer/src/stub.test.ts b/packages/computer/src/stub.test.ts index 6f847ff0c..b8f17469d 100644 --- a/packages/computer/src/stub.test.ts +++ b/packages/computer/src/stub.test.ts @@ -13,7 +13,7 @@ import { enableStubTracking, stubSnapshot } from "@cloudflare/computer-rpc/debug"; import { SQLiteTestStorage } from "@cloudflare/dofs/testing"; -import { beforeAll, describe, expect, it } from "vitest"; +import { beforeAll, describe, expect, it, vi } from "vitest"; import type { BackendHandle, WorkspaceBackend } from "./backend.js"; import { createGitClient } from "./git/index.js"; @@ -357,6 +357,17 @@ 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 }; + vi.spyOn(ws.runtime, "pending").mockResolvedValue([call]); + const respond = vi.spyOn(ws.runtime, "respond").mockResolvedValue(); + await expect(ws.stub().runtime.pending()).resolves.toEqual([call]); + await ws.stub().runtime.respond("run", "run:1", { ok: true }, { backend: "b" }); + expect(respond).toHaveBeenCalledWith("run", "run:1", { ok: true }, { backend: "b" }); + }); + }); + it("runtime.exec rejects when a backend ignores the requested id", async () => { const shellRpc: import("@cloudflare/computer-rpc").ShellRPC = { async exec() { diff --git a/packages/computer/src/stub.ts b/packages/computer/src/stub.ts index a49ad8ed2..90273db58 100644 --- a/packages/computer/src/stub.ts +++ b/packages/computer/src/stub.ts @@ -75,7 +75,9 @@ import type { WorkspaceRuntimeExecOptions, WorkspaceRuntimeGetOptions, WorkspaceRuntimeKillOptions, + WorkspaceRuntimePendingCall, WorkspaceRuntimeResult, + WorkspaceRuntimeValue, } from "./runtime/types.js"; import { encodeRuntimeEvent } from "./runtime/wire.js"; import type { Workspace } from "./workspace.js"; @@ -470,6 +472,20 @@ export class WorkspaceRuntimeStub extends RpcTarget { disposeExec(id: string, options: { backend?: string } = {}): Promise { return this.#ws.runtime.disposeExec(id, options); } + + // Plain data both ways, so nothing here needs disposing. + pending(): Promise { + return this.#ws.runtime.pending(); + } + + respond( + id: string, + callId: string, + value: WorkspaceRuntimeValue, + options: { backend?: string } = {}, + ): Promise { + return this.#ws.runtime.respond(id, callId, value, options); + } } // Assets half. Pure value returns: `publish(path, { expiresAfter }) From 0f9748477f7c22a497ace9bc89b8898b4d6497cd Mon Sep 17 00:00:00 2001 From: scuffi Date: Tue, 20 Oct 2026 11:53:09 +0100 Subject: [PATCH 08/12] computer: Release a run once its calls in flight settle or wait Releasing a run for an interrupt waited for every call that was in flight when the first one interrupted. A call among them that went on to interrupt itself never settled, so the run was never released, its answers never resumed it, and it sat until its timeout failed it. Two host module calls in one Promise.all that each await anything before interrupting did this. quiesce() now waits until each call in flight has been answered or is waiting too, looking again whenever another call starts waiting, so those waits are answered together. --- docs/17_isolate_javascript.md | 2 +- packages/computer/src/runtime/bridge.ts | 21 ++++++++++++++----- .../computer/tests/script-runner-worker.ts | 6 ++++++ packages/computer/tests/script-runner.test.ts | 16 ++++++++++++++ 4 files changed, 39 insertions(+), 6 deletions(-) diff --git a/docs/17_isolate_javascript.md b/docs/17_isolate_javascript.md index d3b0c671d..91019f825 100644 --- a/docs/17_isolate_javascript.md +++ b/docs/17_isolate_javascript.md @@ -225,7 +225,7 @@ Once none of its calls are waiting, the run replays to the call and the function - A call can interrupt once. A second `interrupt()` in the same call throws. - Only a run with `onInterrupt: "replay"` can wait. Anywhere else, `interrupt()` throws, so a Workspace with modules that interrupt should make `"replay"` its backend's default. -- Calls that interrupt together are answered one after another. A run is released with the calls that were in flight, and a call it makes after that waits once it resumes. +- A run is released once each call in flight has finished or is waiting too, and those waits are answered together. A call the run makes after the first interrupt is held until it resumes, so a call that interrupts straight away can leave the next one to wait for a later round. - `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. diff --git a/packages/computer/src/runtime/bridge.ts b/packages/computer/src/runtime/bridge.ts index 0e8b8a375..6d5317831 100644 --- a/packages/computer/src/runtime/bridge.ts +++ b/packages/computer/src/runtime/bridge.ts @@ -35,6 +35,8 @@ export class WorkspaceRuntimeBridge extends RpcTarget { readonly #onWait?: (call: { seq: number; payload: unknown; since: number }) => void; // Set once a call interrupts: new calls are held, never started. #holding = false; + // Wakes quiesce() when another call starts waiting. + #waited?: () => void; #cancelled = false; #callTimedOut = false; #calls = 0; @@ -356,6 +358,7 @@ export class WorkspaceRuntimeBridge extends RpcTarget { return Promise.reject(error); } this.#waiting.add(abort); + this.#waited?.(); this.#onWait?.({ seq: step.seq, payload, since }); } // Settles only when the run is released and the call aborted. @@ -365,14 +368,22 @@ export class WorkspaceRuntimeBridge extends RpcTarget { } /** - * Hold new calls and wait for the ones in flight to be answered and - * journaled, apart from those that interrupted. The run can then be - * released without any call running twice when it resumes. + * Hold new calls and wait until each one in flight has been answered + * and journaled, or is waiting itself. The run can then be released + * without any call running twice when it resumes. */ async quiesce(): Promise { this.#holding = true; - const answering = [...this.#responses].filter(([, abort]) => !this.#waiting.has(abort)); - await Promise.allSettled(answering.map(([response]) => response)); + while (true) { + const answering = [...this.#responses].filter(([, abort]) => !this.#waiting.has(abort)); + if (answering.length === 0) return; + // Look again once one is answered or starts waiting. + await new Promise((resolve) => { + this.#waited = resolve; + const wake = () => resolve(); + void Promise.race(answering.map(([response]) => response)).then(wake, wake); + }); + } } // The isolate is waiting for a journaled call its replay never made. diff --git a/packages/computer/tests/script-runner-worker.ts b/packages/computer/tests/script-runner-worker.ts index b75767190..b6db9afba 100644 --- a/packages/computer/tests/script-runner-worker.ts +++ b/packages/computer/tests/script-runner-worker.ts @@ -177,6 +177,12 @@ function javascriptBackend( countCall(`answered:${action}`); return answer; }, + // Like approve, but awaits first, as a real lookup would. + async approveLater(args, ctx) { + const action = String(args[0]); + await new Promise((resolve) => setTimeout(resolve, 50)); + return ctx.interrupt({ kind: "approval", action }); + }, async marker() { return { __workspace_codec__: { version: 1, type: "bytes", data: [1] }, diff --git a/packages/computer/tests/script-runner.test.ts b/packages/computer/tests/script-runner.test.ts index 7973cceae..44eb96774 100644 --- a/packages/computer/tests/script-runner.test.ts +++ b/packages/computer/tests/script-runner.test.ts @@ -331,6 +331,7 @@ describe("WorkspaceRuntime", () => { expect(response.status, text).toBe(200); expect(JSON.parse(text).result.value).toEqual([ "approve", + "approveLater", "delete", "echo", "gate", @@ -1474,6 +1475,21 @@ describe("interrupts", () => { }); }); + it("waits on calls in flight that interrupt after the first", async () => { + await start( + "wait-later", + ` + import { approveLater } from "ws:test-host"; + export default async () => Promise.all([approveLater("one"), approveLater("two")]); + `, + ); + const first = await waitingCall("wait-later"); + const second = await waitingCall("wait-later", first.callId); + await respond("wait-later", first.callId, 1); + await respond("wait-later", second.callId, 2); + expect(await result("wait-later")).toMatchObject({ status: "completed", value: [1, 2] }); + }); + it("lets other calls in flight finish before it releases the run", async () => { await start( "wait-inflight", From 8513ed71a96045f7b1298daefab1923bf8155afa Mon Sep 17 00:00:00 2001 From: scuffi Date: Tue, 20 Oct 2026 11:53:09 +0100 Subject: [PATCH 09/12] computer: Test that an answer starts a run's attempts again A run resumed after its last wait is answered records its first attempt, since the answer is progress. Without that, a run whose earlier interruptions had used up its attempts could fail on the next one before the answered call was journaled. --- .../worker-javascript.test.ts | 22 +++++++++++++++++++ 1 file changed, 22 insertions(+) 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 7c87460cc..5660a9b22 100644 --- a/packages/computer/src/backends/worker-javascript/worker-javascript.test.ts +++ b/packages/computer/src/backends/worker-javascript/worker-javascript.test.ts @@ -2086,6 +2086,28 @@ describe("WorkerJavaScriptBackend interrupted executions", () => { for (const workspace of [...workspaces, last]) await workspace.close(); }); + it("starts a run's attempts again once its wait is answered", async () => { + const storage = new SQLiteTestStorage(); + const db = new Database(storage); + const workspaces = [await interrupt(storage)]; + for (let attempt = 0; attempt < 2; attempt += 1) { + const workspace = open(storage); + await workspace.runtime.recover(); + workspaces.push(workspace); + } + expect(loadDurable(db, "worker-javascript", "run")?.attempt).toBe(3); + // The run's next call waits for an answer. + const journal = new ExecutionJournal(db, "worker-javascript", "run"); + const step = journal.begin("host/ws:test.approve", []); + if (step.kind !== "live") throw new Error("expected a live call"); + journal.interrupt(step.seq, { kind: "approval" }); + const answering = open(storage); + workspaces.push(answering); + await answering.runtime.respond("run", `run:${step.seq}`, true); + expect(loadDurable(db, "worker-javascript", "run")?.attempt).toBe(1); + for (const workspace of workspaces) await workspace.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, From 6e01fd93490ecd05f7c074963a506906dcd92cff Mon Sep 17 00:00:00 2001 From: scuffi Date: Tue, 20 Oct 2026 11:53:09 +0100 Subject: [PATCH 10/12] computer: Merge the journal's two interrupt() doc comments --- .../computer/src/backends/worker-javascript/journal.ts | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/packages/computer/src/backends/worker-javascript/journal.ts b/packages/computer/src/backends/worker-javascript/journal.ts index f4b8b39fc..ac3be0dc8 100644 --- a/packages/computer/src/backends/worker-javascript/journal.ts +++ b/packages/computer/src/backends/worker-javascript/journal.ts @@ -607,11 +607,10 @@ export class ExecutionJournal implements BridgeJournal { } /** - * Mark a live call as waiting for an answer. Its response is never - * committed: the run is released, and the call runs again once - * answered. + * Mark a live call as waiting for an answer, and return when its run + * started waiting. Its response is never committed: the run is + * released, and the call runs again once answered. */ - /** Mark a call waiting, and return when its run started waiting. */ interrupt(seq: number, payload: unknown): number { this.#db.run( `UPDATE workspace_runtime_journal SET state = 'waiting', payload = ? From 8e9261b64563f0ab047a481c393964aae765a034 Mon Sep 17 00:00:00 2001 From: scuffi Date: Wed, 21 Oct 2026 15:18:25 +0100 Subject: [PATCH 11/12] computer: Report when each pending call started waiting pending() and onWait gave every waiting call its run's waiting_since, which is when the run's first unanswered call started waiting, so a second call that started waiting later showed the first one's time. The journal now keeps waiting_since for each call, and both report that. The run's own waiting_since still decides how much waiting its timeout leaves out. --- .../worker-javascript/journal.test.ts | 25 ++++++++++++++++ .../src/backends/worker-javascript/journal.ts | 30 ++++++++++--------- packages/computer/src/runtime/bridge.ts | 2 +- packages/computer/src/runtime/types.ts | 2 +- 4 files changed, 43 insertions(+), 16 deletions(-) diff --git a/packages/computer/src/backends/worker-javascript/journal.test.ts b/packages/computer/src/backends/worker-javascript/journal.test.ts index 84df4fafb..3728394db 100644 --- a/packages/computer/src/backends/worker-javascript/journal.test.ts +++ b/packages/computer/src/backends/worker-javascript/journal.test.ts @@ -134,6 +134,31 @@ describe("ExecutionJournal", () => { }); }); + it("keeps when each call started waiting, and the run's first", () => { + const db = database(); + db.run( + `INSERT INTO workspace_runtime_resume (backend, execution_id, on_interrupt, attempt) + VALUES ('js', 'run', 'replay', 1)`, + ); + const journal = new ExecutionJournal(db, "js", "run"); + const now = vi.spyOn(Date, "now"); + for (const [action, at] of [ + ["one", 1000], + ["two", 5000], + ] as const) { + const step = journal.begin("host/ws:test.approve", [action]); + if (step.kind !== "live") throw new Error("expected a live call"); + now.mockReturnValue(at); + expect(journal.interrupt(step.seq, { action })).toBe(at); + } + now.mockRestore(); + expect(waitingCalls(db, "js").map((call) => call.since)).toEqual([1000, 5000]); + expect( + db.one<{ waiting_since: number }>("SELECT waiting_since FROM workspace_runtime_resume") + ?.waiting_since, + ).toBe(1000); + }); + it("numbers responses in the order they were committed", () => { const db = database(); const first = new ExecutionJournal(db, "js", "run"); diff --git a/packages/computer/src/backends/worker-javascript/journal.ts b/packages/computer/src/backends/worker-javascript/journal.ts index 63a2e4e51..3d6211b5a 100644 --- a/packages/computer/src/backends/worker-javascript/journal.ts +++ b/packages/computer/src/backends/worker-javascript/journal.ts @@ -172,6 +172,7 @@ export function createResumeTables(db: Database): void { state TEXT NOT NULL, response TEXT, interrupt_payload TEXT, + waiting_since INTEGER, answer TEXT, journaled_at INTEGER, journal_order INTEGER, @@ -329,12 +330,10 @@ export function answerCall( export function waitingCalls(db: Database, backend: string): WaitingCall[] { return db .all<{ execution_id: string; seq: number; interrupt_payload: string; waiting_since: number }>( - `SELECT j.execution_id, j.seq, j.interrupt_payload, d.waiting_since - FROM workspace_runtime_journal j - JOIN workspace_runtime_resume d - ON d.backend = j.backend AND d.execution_id = j.execution_id - WHERE j.backend = ? AND j.state = 'waiting' - ORDER BY d.waiting_since, j.execution_id, j.seq`, + `SELECT execution_id, seq, interrupt_payload, waiting_since + FROM workspace_runtime_journal + WHERE backend = ? AND state = 'waiting' + ORDER BY waiting_since, execution_id, seq`, backend, ) .map((row) => ({ @@ -610,28 +609,31 @@ export class ExecutionJournal implements BridgeJournal { } /** - * Mark a live call as waiting for an answer, and return when its run + * Mark a live call as waiting for an answer, and return when it * started waiting. Its response is never committed: the run is - * released, and the call runs again once answered. + * released, and the call runs again once answered. The run's own + * waiting_since is when its first unanswered call started waiting, + * which is what its timeout leaves out. */ interrupt(seq: number, payload: unknown): number { + const now = Date.now(); this.#db.run( - `UPDATE workspace_runtime_journal SET state = 'waiting', interrupt_payload = ? + `UPDATE workspace_runtime_journal SET state = 'waiting', interrupt_payload = ?, waiting_since = ? WHERE backend = ? AND execution_id = ? AND seq = ?`, encodeLimited(payload, "payload"), + now, this.#backend, this.#id, seq, ); - const [row] = this.#db.all<{ waiting_since: number }>( + this.#db.run( `UPDATE workspace_runtime_resume SET waiting_since = COALESCE(waiting_since, ?) - WHERE backend = ? AND execution_id = ? - RETURNING waiting_since`, - Date.now(), + WHERE backend = ? AND execution_id = ?`, + now, this.#backend, this.#id, ); - return Number(row?.waiting_since ?? Date.now()); + return now; } diverged(message: string): void { diff --git a/packages/computer/src/runtime/bridge.ts b/packages/computer/src/runtime/bridge.ts index 6d5317831..c1828438f 100644 --- a/packages/computer/src/runtime/bridge.ts +++ b/packages/computer/src/runtime/bridge.ts @@ -632,7 +632,7 @@ export interface BridgeJournal { | { readonly kind: "diverge"; readonly message: string }; /** Record a live call's response, and return its time and ordinal. */ commit(seq: number, response: BridgeResponse): { at: number; ordinal: number }; - /** Mark a live call as waiting for an answer. */ + /** Mark a live call as waiting for an answer, and return when it started. */ interrupt(seq: number, payload: unknown): number; /** Save a chunk of the isolate's event log. Throws on a malformed one. */ record(position: unknown, events: unknown): void; diff --git a/packages/computer/src/runtime/types.ts b/packages/computer/src/runtime/types.ts index 18c7fbc19..f795cbc74 100644 --- a/packages/computer/src/runtime/types.ts +++ b/packages/computer/src/runtime/types.ts @@ -317,7 +317,7 @@ export interface WorkspaceRuntimePendingCall { readonly callId: string; /** What the call passed to `ctx.interrupt()`. */ readonly payload: WorkspaceRuntimeValue; - /** When the run started waiting, in milliseconds since the epoch. */ + /** When the call started waiting, in milliseconds since the epoch. */ readonly since: number; } From ba58d1e087cc6f9bd6bdd18c67eadb0a21ffff8f Mon Sep 17 00:00:00 2001 From: scuffi Date: Wed, 21 Oct 2026 15:18:26 +0100 Subject: [PATCH 12/12] computer: Call respond()'s value its answer The docs, the journal column and the ENOTWAITING message all call what respond() is given an answer, so the parameter says so too. It's positional, so callers don't change. --- .../src/backends/worker-javascript/worker-javascript.ts | 6 +++--- packages/computer/src/runtime/runtime.ts | 6 +++--- packages/computer/src/runtime/types.ts | 2 +- packages/computer/src/stub.ts | 4 ++-- 4 files changed, 9 insertions(+), 9 deletions(-) diff --git a/packages/computer/src/backends/worker-javascript/worker-javascript.ts b/packages/computer/src/backends/worker-javascript/worker-javascript.ts index 4fc80bf5c..26262a5e3 100644 --- a/packages/computer/src/backends/worker-javascript/worker-javascript.ts +++ b/packages/computer/src/backends/worker-javascript/worker-javascript.ts @@ -633,16 +633,16 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { })); } - async respond(input: { id: string; callId: string; value: WorkspaceRuntimeValue }) { + async respond(input: { id: string; callId: string; answer: WorkspaceRuntimeValue }) { assertExecutionId(input.id); - assertRuntimeValue(input.value); + assertRuntimeValue(input.answer); const record = this.#recordFor(input.id); const prefix = `${input.id}:`; const seq = input.callId.startsWith(prefix) ? Number(input.callId.slice(prefix.length)) : NaN; const waiting = record?.status === "running" && Number.isSafeInteger(seq) ? this.#host.db.transactionSync(() => - answerCall(this.#host.db, this.#backendId, input.id, seq, input.value), + answerCall(this.#host.db, this.#backendId, input.id, seq, input.answer), ) : undefined; if (record === undefined || waiting === undefined) { diff --git a/packages/computer/src/runtime/runtime.ts b/packages/computer/src/runtime/runtime.ts index bc6d7fd25..1a3869ffd 100644 --- a/packages/computer/src/runtime/runtime.ts +++ b/packages/computer/src/runtime/runtime.ts @@ -191,7 +191,7 @@ export class WorkspaceRuntime { /** * Answer a call waiting in `ctx.interrupt()`. The run resumes once - * none of its calls are still waiting, and the call gets `value` as + * none of its calls are still waiting, and the call gets `answer` as * what `interrupt()` returns. Throws `ENOTWAITING` if the call isn't * waiting, for example because it was already answered or the run * was killed. Experimental: it may change in any release. @@ -199,7 +199,7 @@ export class WorkspaceRuntime { async respond( id: string, callId: string, - value: WorkspaceRuntimeValue, + answer: WorkspaceRuntimeValue, options: { backend?: string } = {}, ): Promise { assertExecutionId(id); @@ -208,7 +208,7 @@ export class WorkspaceRuntime { if (handle.respond === undefined) { throw new Error(`Backend ${backend} doesn't run calls that wait for an answer.`); } - await handle.respond({ id, callId, value }); + await handle.respond({ id, callId, answer }); } async killExec(id: string, options: WorkspaceRuntimeKillOptions = {}): Promise { diff --git a/packages/computer/src/runtime/types.ts b/packages/computer/src/runtime/types.ts index f795cbc74..7c898b883 100644 --- a/packages/computer/src/runtime/types.ts +++ b/packages/computer/src/runtime/types.ts @@ -353,7 +353,7 @@ export interface WorkspaceModuleBackendHandle { recover?(): Promise; // Calls waiting for an answer, and answering one. pending?(): Promise[]>; - respond?(input: { id: string; callId: string; value: WorkspaceRuntimeValue }): Promise; + respond?(input: { id: string; callId: string; answer: WorkspaceRuntimeValue }): Promise; // Tear down a backend-owned transport. The command adapter omits // it: a command backend's transport is closed through its // BackendHandle, not through the adapter the runtime consumes. diff --git a/packages/computer/src/stub.ts b/packages/computer/src/stub.ts index 90273db58..58dcf505b 100644 --- a/packages/computer/src/stub.ts +++ b/packages/computer/src/stub.ts @@ -481,10 +481,10 @@ export class WorkspaceRuntimeStub extends RpcTarget { respond( id: string, callId: string, - value: WorkspaceRuntimeValue, + answer: WorkspaceRuntimeValue, options: { backend?: string } = {}, ): Promise { - return this.#ws.runtime.respond(id, callId, value, options); + return this.#ws.runtime.respond(id, callId, answer, options); } }