computer: Let host module calls wait for an answer

.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.

docs/17_isolate_javascript.md

@@ -195,6 +195,40 @@ modules: {
 }
 ```
 
+### Waiting for an answer
+
+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: {
+  "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 });
+}
+```
+
+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.
+
+- 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.
+- 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.
+
 ### Limits
 
 - A run gets three attempts, counted again each time it journals a call, and has to finish within its original timeout, counted from when it first started.

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,
   };
 }

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 |

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,
   createResumeTables,
   decodeValue,
   describeCall,
   encodeValue,
   ExecutionJournal,
+  waitingCalls,
 } from "./journal.js";
 
 function database() {
@@ -97,6 +99,66 @@ 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_resume (backend, execution_id, on_interrupt, 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("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");