From 076edbabee5f1b8e8cca1aebed9c02cc83edb208 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Mon, 20 Jul 2026 01:32:39 -0700 Subject: [PATCH] Suspend sandbox execution deadlines while a tool dispatch awaits the host --- .../suspend-private-sandbox-deadlines.md | 6 + .changeset/suspend-sandbox-deadlines.md | 5 + .../resume-after-sandbox-deadline.test.ts | 150 ++++++++++++++++++ .../core/execution/src/tool-invoker.test.ts | 29 ++-- .../runtime-deno-subprocess/src/index.test.ts | 54 +++++-- .../runtime-deno-subprocess/src/index.ts | 75 +++++---- .../src/invocation.test.ts | 58 +++++++ .../src/module-template.ts | 28 +++- .../kernel/runtime-quickjs/src/index.test.ts | 48 ++++++ packages/kernel/runtime-quickjs/src/index.ts | 93 ++++++++--- 10 files changed, 473 insertions(+), 73 deletions(-) create mode 100644 .changeset/suspend-private-sandbox-deadlines.md create mode 100644 .changeset/suspend-sandbox-deadlines.md create mode 100644 e2e/scenarios/resume-after-sandbox-deadline.test.ts diff --git a/.changeset/suspend-private-sandbox-deadlines.md b/.changeset/suspend-private-sandbox-deadlines.md new file mode 100644 index 000000000..2fb223ee1 --- /dev/null +++ b/.changeset/suspend-private-sandbox-deadlines.md @@ -0,0 +1,6 @@ +--- +"@executor-js/runtime-dynamic-worker": patch +"@executor-js/runtime-deno-subprocess": patch +--- + +Suspend sandbox execution deadlines while tool calls await the host, and reset the autonomous-compute budget after each dispatch returns. diff --git a/.changeset/suspend-sandbox-deadlines.md b/.changeset/suspend-sandbox-deadlines.md new file mode 100644 index 000000000..ff25aa2e2 --- /dev/null +++ b/.changeset/suspend-sandbox-deadlines.md @@ -0,0 +1,5 @@ +--- +"@executor-js/runtime-quickjs": patch +--- + +Suspend sandbox execution deadlines while tool calls await the host, and reset the autonomous-compute budget after each dispatch returns. diff --git a/e2e/scenarios/resume-after-sandbox-deadline.test.ts b/e2e/scenarios/resume-after-sandbox-deadline.test.ts new file mode 100644 index 000000000..42c915442 --- /dev/null +++ b/e2e/scenarios/resume-after-sandbox-deadline.test.ts @@ -0,0 +1,150 @@ +// Cross-target: approval durability against the sandbox clock — the "a human +// is allowed to take a few minutes" promise. A paused execution suspends the +// sandbox mid-call, but the sandbox's execution deadline (5 minutes) used to +// keep ticking while the human decided. Each pause advertises its own +// approval window, so with chained approvals a later window legitimately +// extends past the sandbox's absolute clock: the human answers every window +// on time, yet the fiber is already dead and the final approval lands on an +// unknown execution. +// +// The journey drives exactly that shape: ONE execution with TWO approval +// gates. The first approval is granted late in its window (~3.5 min), so the +// second pause's window reaches well past the sandbox's 5-minute mark. The +// second approval arrives ~5.75 min after execution start — inside its OWN +// advertised window, but past the old absolute deadline. Deliberately slow +// (~6 min): the elapsed time IS the subject under test. A single-pause +// variant cannot express this cross-target — hosts that advertise a +// 4-minute window would expire it legitimately before the sandbox clock +// even matters. +// +// The gate is `policies.create`'s own `requiresApproval` annotation +// (hermetic, same device as policy-tool-approval.test.ts); both approvals +// happen in the same MCP session, so the only variable is elapsed time. +import { randomUUID } from "node:crypto"; + +import { expect } from "@effect/vitest"; +import { Effect } from "effect"; +import { composePluginApi } from "@executor-js/api/server"; + +import { scenario } from "../src/scenario"; +import { Api, Mcp, Target } from "../src/services"; +import { configuredMcpPausedSessionIdleTimeoutMs } from "../setup/mcp-session-timeouts"; + +const coreApi = composePluginApi([] as const); + +// Grant the first approval at 3.5 min — late but inside its 4-minute window. +// The second pause then opens a fresh window reaching ~7.5 min. +const FIRST_APPROVAL_DELAY_MS = 3.5 * 60_000; +// Grant the second approval 2.25 min later: ~5.75 min after execution start, +// past the sandbox's 5-minute budget but inside the second window. +const SECOND_APPROVAL_DELAY_MS = 2.25 * 60_000; + +/** Sandbox code that creates two policies through the approval-gated core + * tool. Patterns are unique-per-run and match no real tool, so the rules are + * inert even if leaked. */ +const createPoliciesCode = (firstPattern: string, secondPattern: string) => ` +const first = await tools.executor.coreTools.policies.create({ + owner: "user", + pattern: ${JSON.stringify(firstPattern)}, + action: "block", +}); +const second = await tools.executor.coreTools.policies.create({ + owner: "user", + pattern: ${JSON.stringify(secondPattern)}, + action: "block", +}); +return JSON.stringify({ first: first.ok, second: second.ok }); +`; + +// The journey spans ~6 real minutes of paused waiting, so the host must keep +// the paused session alive that long. The suite's default e2e override shrinks +// the paused-session idle teardown to seconds (to keep teardown tests fast), +// which would evict the session mid-scenario for reasons unrelated to the +// clock under test — require the production-like window instead. +const PAUSED_IDLE_WINDOW_TOO_SHORT = + configuredMcpPausedSessionIdleTimeoutMs() < 8 * 60_000 + ? `the target's paused-session idle teardown (${configuredMcpPausedSessionIdleTimeoutMs()}ms) evicts the session before this ~6-minute journey completes; boot the target with MCP_PAUSED_SESSION_IDLE_TIMEOUT_MS >= 480000 to run it` + : undefined; + +scenario( + "MCP · chained approvals granted within their windows survive the sandbox clock", + { timeout: 480_000, skip: PAUSED_IDLE_WINDOW_TOO_SHORT }, + Effect.gen(function* () { + const target = yield* Target; + const apiSurface = yield* Api; + const mcp = yield* Mcp; + const identity = yield* target.newIdentity(); + const client = yield* apiSurface.client(coreApi, identity); + const runId = randomUUID().slice(0, 8); + const firstPattern = `resume-deadline-a-${runId}.*`; + const secondPattern = `resume-deadline-b-${runId}.*`; + + const cleanup = client.policies.list().pipe( + Effect.flatMap((list) => + Effect.forEach( + list.filter((p) => p.pattern === firstPattern || p.pattern === secondPattern), + (p) => + client.policies + .remove({ params: { policyId: p.id }, payload: { owner: "user" } }) + .pipe(Effect.ignore), + ), + ), + Effect.ignore, + ); + + yield* Effect.gen(function* () { + const session = mcp.session(identity); + const tools = yield* session.listTools(); + expect(tools).toContain("execute"); + + const paused = yield* session.call("execute", { + code: createPoliciesCode(firstPattern, secondPattern), + }); + expect(paused.text, "the first gated call paused for approval").toContain("Execution paused"); + const firstMatch = /\bexecutionId:\s*(\S+)/.exec(paused.text); + expect(firstMatch, "the first pause carries an executionId").not.toBeNull(); + + // The human takes most of the first approval window before deciding. + yield* Effect.sleep(`${FIRST_APPROVAL_DELAY_MS} millis`); + + const secondPaused = yield* session.call("resume", { + executionId: firstMatch![1]!, + action: "accept", + content: JSON.stringify({}), + }); + expect( + secondPaused.text, + "the on-time first approval reaches its pause and surfaces the second gate", + ).toContain("Execution paused"); + const secondMatch = /\bexecutionId:\s*(\S+)/.exec(secondPaused.text); + expect(secondMatch, "the second pause carries an executionId").not.toBeNull(); + + // The second decision lands past the sandbox's 5-minute budget (counted + // from execution start) but well inside the second advertised window. + yield* Effect.sleep(`${SECOND_APPROVAL_DELAY_MS} millis`); + + const resumed = yield* session.call("resume", { + executionId: secondMatch![1]!, + action: "accept", + content: JSON.stringify({}), + }); + + expect( + resumed.text, + "the second on-time approval reaches a live pause instead of a dead or unknown execution", + ).not.toMatch(/Paused execution is unknown|Paused execution expired|timed out after/); + expect(resumed.ok, "the fully approved execution completed without error").toBe(true); + + // Both approvals took effect: the gated tool ran twice. + const afterApproval = yield* client.policies.list(); + expect( + afterApproval.some((p) => p.pattern === firstPattern), + "the first gated tool ran after its approval", + ).toBe(true); + expect( + afterApproval.some((p) => p.pattern === secondPattern), + "the second gated tool ran after the late-but-in-window approval", + ).toBe(true); + }).pipe(Effect.ensuring(cleanup)); + }), +); diff --git a/packages/core/execution/src/tool-invoker.test.ts b/packages/core/execution/src/tool-invoker.test.ts index 0a8018b0d..dd25e6168 100644 --- a/packages/core/execution/src/tool-invoker.test.ts +++ b/packages/core/execution/src/tool-invoker.test.ts @@ -1743,28 +1743,37 @@ describe("pause/resume with multiple elicitations", () => { { timeout: 10000 }, ); - // live clock: the sandbox timeout is a real timer, so the test must - // actually wait for it rather than suspend on the virtual TestClock. + // Live clock: the failing executor delays long enough for its detached tool + // invocation to publish the pause before the executor settles on its own. it.live( "a pause abandoned by a failing sandbox is dropped and its resume replays the failure outcome", () => Effect.gen(function* () { const executor = yield* makeElicitingExecutor(); - // Sandbox times out while suspended on the elicitation, so the fiber - // settles without a resume ever arriving. + const failingCodeExecutor = { + execute: (_code, toolInvoker) => + Effect.gen(function* () { + yield* Effect.forkChild( + toolInvoker + .invoke({ path: "api.org.main.singleApproval", args: {} }) + .pipe(Effect.ignore), + ); + yield* Effect.sleep("100 millis"); + return { result: null, error: "sandbox failed after pausing" }; + }), + } satisfies typeof codeExecutor; const engine = createExecutionEngine({ executor, - codeExecutor: makeQuickJsExecutor({ timeoutMs: 250 }), + codeExecutor: failingCodeExecutor, }); - const code = "return await tools.api.org.main.singleApproval({});"; - const outcome1 = yield* engine.executeWithPause(code); + const outcome1 = yield* engine.executeWithPause("ignored"); expect(outcome1.status).toBe("paused"); if (outcome1.status !== "paused") return; const executionId = outcome1.execution.id; - // Wait for the sandbox timeout to settle the detached fiber. - yield* Effect.sleep("600 millis"); + // Wait for the sandbox failure to settle the detached fiber. + yield* Effect.sleep("300 millis"); // The dead pause must no longer be reported as live... const stillPaused = yield* engine.getPausedExecution(executionId); @@ -1774,7 +1783,7 @@ describe("pause/resume with multiple elicitations", () => { const late = yield* engine.resume(executionId, { action: "accept" }); expect(late?.status).toBe("completed"); const completed = late as Extract, { status: "completed" }>; - expect(completed.result.error).toContain("timed out"); + expect(completed.result.error).toBe("sandbox failed after pausing"); }), { timeout: 10000 }, ); diff --git a/packages/kernel/runtime-deno-subprocess/src/index.test.ts b/packages/kernel/runtime-deno-subprocess/src/index.test.ts index 92f4aba38..aadcedcb0 100644 --- a/packages/kernel/runtime-deno-subprocess/src/index.test.ts +++ b/packages/kernel/runtime-deno-subprocess/src/index.test.ts @@ -234,19 +234,51 @@ describe.skipIf(!isDenoAvailable())("runtime-deno-subprocess", () => { }), ); - it.effect("respects timeout", () => - Effect.gen(function* () { - const executor = makeDenoSubprocessExecutor({ - timeoutMs: 500, - }); - const toolInvoker = makeTestInvoker({}); + it("suspends the execution deadline while a tool dispatch is in flight", async () => { + const timeoutMs = 300; + const executor = makeDenoSubprocessExecutor({ timeoutMs }); + const toolInvoker: SandboxToolInvoker = { + invoke: () => Effect.sleep(timeoutMs * 3).pipe(Effect.as("slow result")), + }; + + const output = await Effect.runPromise( + executor.execute("return await tools.slow.wait({});", toolInvoker), + ); + + expect(output.error).toBeUndefined(); + expect(output.result).toBe("slow result"); + }); + + it("still times out continuous autonomous compute", async () => { + const timeoutMs = 300; + const executor = makeDenoSubprocessExecutor({ timeoutMs }); + const toolInvoker = makeTestInvoker({}); - const output = yield* executor.execute("await new Promise(() => {}); return 1;", toolInvoker); + const output = await Effect.runPromise( + executor.execute("await new Promise(() => {}); return 1;", toolInvoker), + ); - expect(output.result).toBeNull(); - expect(output.error).toContain("timed out"); - }), - ); + expect(output.result).toBeNull(); + expect(output.error).toBe(`Deno subprocess execution timed out after ${timeoutMs}ms`); + }); + + it("resets the execution deadline after a tool dispatch returns", async () => { + const timeoutMs = 300; + const executor = makeDenoSubprocessExecutor({ timeoutMs }); + const toolInvoker: SandboxToolInvoker = { + invoke: () => Effect.sleep(timeoutMs * 3).pipe(Effect.as("done")), + }; + + const output = await Effect.runPromise( + executor.execute( + "await tools.slow.wait({}); await new Promise(() => {}); return 1;", + toolInvoker, + ), + ); + + expect(output.result).toBeNull(); + expect(output.error).toBe(`Deno subprocess execution timed out after ${timeoutMs}ms`); + }); it.effect("network access is denied by default", () => Effect.gen(function* () { diff --git a/packages/kernel/runtime-deno-subprocess/src/index.ts b/packages/kernel/runtime-deno-subprocess/src/index.ts index caf20ccaa..8ef4e78ad 100644 --- a/packages/kernel/runtime-deno-subprocess/src/index.ts +++ b/packages/kernel/runtime-deno-subprocess/src/index.ts @@ -230,11 +230,21 @@ const executeInDeno = ( catch: (cause) => new DenoSpawnError({ executable: denoExecutable, reason: cause }), }); + const start = Date.now(); + let inFlight = 0; + let lastReturnedAt = start; + // Send code to the subprocess writeMessage(worker.stdin, { type: "start", code: recoveredBody, nonce }); - // Set up timeout — kills process and completes the deferred - const timer = setTimeout(() => { + // Measure only continuous autonomous subprocess compute. Tool dispatches + // suspend the clock and each return grants a fresh timeout budget. + const timer = setInterval(() => { + const now = Date.now(); + if (inFlight > 0 || now - Math.max(start, lastReturnedAt) < timeoutMs) { + return; + } + worker.dispose(); runSync( completeWith({ @@ -242,7 +252,7 @@ const executeInDeno = ( error: `Deno subprocess execution timed out after ${timeoutMs}ms`, }), ); - }, timeoutMs); + }, 1_000); // ----------------------------------------------------------------------- // Message processing fiber — tool calls happen here, inside Effect @@ -255,30 +265,39 @@ const executeInDeno = ( switch (msg.type) { case "tool_call": { - const toolResult = yield* toolInvoker - .invoke({ path: msg.toolPath, args: msg.args }) - .pipe( - Effect.map( - (value): HostToWorkerMessage => ({ - type: "tool_result", - nonce, - requestId: msg.requestId, - ok: true, - value, - }), - ), - Effect.catchCause((cause) => - Effect.succeed({ - type: "tool_result", - nonce, - requestId: msg.requestId, - ok: false, - error: causeMessage(cause), - }), - ), - ); - - writeMessage(worker.stdin, toolResult); + // Symmetric bracket (mirrors ToolDispatcher.call): the decrement + // must run even if the invoke or write dies, or the watchdog's + // in-flight guard would suspend the timeout forever. + inFlight += 1; + yield* toolInvoker.invoke({ path: msg.toolPath, args: msg.args }).pipe( + Effect.map( + (value): HostToWorkerMessage => ({ + type: "tool_result", + nonce, + requestId: msg.requestId, + ok: true, + value, + }), + ), + Effect.catchCause((cause) => + Effect.succeed({ + type: "tool_result", + nonce, + requestId: msg.requestId, + ok: false, + error: causeMessage(cause), + }), + ), + Effect.flatMap((toolResult) => + Effect.sync(() => writeMessage(worker.stdin, toolResult)), + ), + Effect.ensuring( + Effect.sync(() => { + inFlight -= 1; + lastReturnedAt = Date.now(); + }), + ), + ); break; } @@ -318,7 +337,7 @@ const executeInDeno = ( const output = yield* Deferred.await(result).pipe( Effect.ensuring( Effect.gen(function* () { - clearTimeout(timer); + clearInterval(timer); yield* Fiber.interrupt(processFiber); worker.dispose(); }), diff --git a/packages/kernel/runtime-dynamic-worker/src/invocation.test.ts b/packages/kernel/runtime-dynamic-worker/src/invocation.test.ts index b54869a8f..336866780 100644 --- a/packages/kernel/runtime-dynamic-worker/src/invocation.test.ts +++ b/packages/kernel/runtime-dynamic-worker/src/invocation.test.ts @@ -484,6 +484,64 @@ describe("makeDynamicWorkerExecutor", () => { expect(result.result).toBe(30); }); + it("suspends the execution deadline while a tool dispatch is in flight", async () => { + const timeoutMs = 200; + const executor = makeDynamicWorkerExecutor({ + loader, + timeoutMs, + hostTimeoutGraceMs: 5_000, + }); + const invoker: SandboxToolInvoker = { + invoke: () => Effect.sleep(timeoutMs * 3).pipe(Effect.as("slow result")), + }; + + const result = await Effect.runPromise( + executor.execute("async () => await tools.slow.wait({})", invoker), + ); + + expect(result.error).toBeUndefined(); + expect(result.result).toBe("slow result"); + }); + + it("still times out continuous autonomous compute", async () => { + const timeoutMs = 200; + const executor = makeDynamicWorkerExecutor({ + loader, + timeoutMs, + hostTimeoutGraceMs: 5_000, + }); + const invoker = makeInvoker(() => null); + + const result = await Effect.runPromise( + executor.execute("async () => await new Promise(() => {})", invoker), + ); + + expect(result.result).toBeNull(); + expect(result.error).toBe(`Execution timed out after ${timeoutMs}ms`); + }); + + it("resets the execution deadline after a tool dispatch returns", async () => { + const timeoutMs = 200; + const executor = makeDynamicWorkerExecutor({ + loader, + timeoutMs, + hostTimeoutGraceMs: 5_000, + }); + const invoker: SandboxToolInvoker = { + invoke: () => Effect.sleep(timeoutMs * 3).pipe(Effect.as("done")), + }; + + const result = await Effect.runPromise( + executor.execute( + "async () => { await tools.slow.wait({}); await new Promise(() => {}); }", + invoker, + ), + ); + + expect(result.result).toBeNull(); + expect(result.error).toBe(`Execution timed out after ${timeoutMs}ms`); + }); + it("returns an execution error for circular tool args", async () => { const executor = makeDynamicWorkerExecutor({ loader }); const invoker = makeInvoker(() => null); diff --git a/packages/kernel/runtime-dynamic-worker/src/module-template.ts b/packages/kernel/runtime-dynamic-worker/src/module-template.ts index 38c8c67ee..199a2c14c 100644 --- a/packages/kernel/runtime-dynamic-worker/src/module-template.ts +++ b/packages/kernel/runtime-dynamic-worker/src/module-template.ts @@ -22,6 +22,9 @@ export const buildExecutorModule = (body: string, timeoutMs: number): string => "", "export default class CodeExecutor extends WorkerEntrypoint {", " async evaluate(__dispatcher) {", + " const __start = Date.now();", + " let __inFlight = 0;", + " let __lastReturnedAt = __start;", " const __logs = [];", " const __outputs = [];", ' console.log = (...a) => { __logs.push(a.map(String).join(" ")); };', @@ -196,7 +199,14 @@ export const buildExecutorModule = (body: string, timeoutMs: number): string => " if (!toolPath) throw new Error('Tool path missing in invocation');", " return (async () => {", " const encoded = await __encodeBinary(args[0]);", - " const data = await __dispatcher.call(toolPath, encoded);", + " __inFlight += 1;", + " let data;", + " try {", + " data = await __dispatcher.call(toolPath, encoded);", + " } finally {", + " __inFlight -= 1;", + " __lastReturnedAt = Date.now();", + " }", " if (!data.ok) throw new Error(__publicToolErrorMessage(data.error) || 'Internal tool error');", " return __decodeBinary(data.result);", " })();", @@ -204,18 +214,28 @@ export const buildExecutorModule = (body: string, timeoutMs: number): string => " });", " const tools = __makeToolsProxy();", "", + " let __watchdogInterval;", + " const __watchdog = new Promise((_, reject) => {", + " __watchdogInterval = setInterval(() => {", + " const now = Date.now();", + ` if (__inFlight === 0 && now - Math.max(__start, __lastReturnedAt) >= ${timeoutMs}) {`, + ` reject(new Error("Execution timed out after ${timeoutMs}ms"));`, + " }", + " }, 1000);", + " });", + "", " try {", " const result = await Promise.race([", " (async () => {", body, " })(),", - " new Promise((_, reject) =>", - ` setTimeout(() => reject(new Error("Execution timed out after ${timeoutMs}ms")), ${timeoutMs})`, - " ),", + " __watchdog,", " ]);", " return { result, output: __outputs.length > 0 ? __outputs : undefined, logs: __logs };", " } catch (err) {", " return { result: undefined, output: __outputs.length > 0 ? __outputs : undefined, error: __serializeThrownError(err), logs: __logs };", + " } finally {", + " clearInterval(__watchdogInterval);", " }", " }", "}", diff --git a/packages/kernel/runtime-quickjs/src/index.test.ts b/packages/kernel/runtime-quickjs/src/index.test.ts index ae1145bfa..a55f022ac 100644 --- a/packages/kernel/runtime-quickjs/src/index.test.ts +++ b/packages/kernel/runtime-quickjs/src/index.test.ts @@ -150,6 +150,54 @@ describe("quickjs executor", () => { }), ); + it("suspends the execution deadline while a tool dispatch is in flight", async () => { + const timeoutMs = 100; + const slowInvoker: SandboxToolInvoker = { + invoke: () => Effect.sleep(timeoutMs * 3).pipe(Effect.as("slow result")), + }; + const slowExecutor = makeQuickJsExecutor({ timeoutMs }); + + const result = await Effect.runPromise( + slowExecutor.execute("return await tools.slow.wait({});", slowInvoker), + ); + + expect(result.error).toBeUndefined(); + expect(result.result).toBe("slow result"); + }); + + it("still times out continuous autonomous compute", async () => { + const timeoutMs = 100; + const timedExecutor = makeQuickJsExecutor({ timeoutMs }); + + const result = await Effect.runPromise( + timedExecutor.execute("while (true) {}", makeTestInvoker({})), + ); + + expect(result.result).toBeNull(); + expect(result.error).toBe(`QuickJS execution timed out after ${timeoutMs}ms`); + }); + + it("resets the execution deadline after a tool dispatch returns", async () => { + const timeoutMs = 100; + const slowInvoker: SandboxToolInvoker = { + invoke: () => Effect.sleep(timeoutMs * 3).pipe(Effect.as("done")), + }; + const timedExecutor = makeQuickJsExecutor({ timeoutMs }); + + const result = await Effect.runPromise( + timedExecutor.execute( + ` + await tools.slow.wait({}); + while (true) {} + `, + slowInvoker, + ), + ); + + expect(result.result).toBeNull(); + expect(result.error).toBe(`QuickJS execution timed out after ${timeoutMs}ms`); + }); + it.effect("invokes multiple tools in sequence", () => Effect.gen(function* () { const invoker = makeTestInvoker({ diff --git a/packages/kernel/runtime-quickjs/src/index.ts b/packages/kernel/runtime-quickjs/src/index.ts index 2ccebbb25..93341c9aa 100644 --- a/packages/kernel/runtime-quickjs/src/index.ts +++ b/packages/kernel/runtime-quickjs/src/index.ts @@ -10,7 +10,6 @@ import * as Data from "effect/Data"; import * as Effect from "effect/Effect"; import { getQuickJS, - shouldInterruptAfterDeadline, type QuickJSContext, type QuickJSDeferredPromise, type QuickJSHandle, @@ -109,9 +108,45 @@ const looksLikeInterruptedError = (message: string): boolean => /\binterrupted\b const timeoutMessage = (timeoutMs: number): string => `QuickJS execution timed out after ${timeoutMs}ms`; -const normalizeExecutionError = (cause: unknown, deadlineMs: number, timeoutMs: number): string => { +type DeadlineTracker = { + /** Current absolute deadline, or null while at least one dispatch is in flight. */ + readonly deadlineMs: () => number | null; + readonly dispatchStarted: () => void; + readonly dispatchReturned: () => void; +}; + +const makeDeadlineTracker = (timeoutMs: number): DeadlineTracker => { + const start = Date.now(); + let inFlight = 0; + let lastReturnedAt = start; + + return { + deadlineMs: () => (inFlight > 0 ? null : Math.max(start, lastReturnedAt) + timeoutMs), + dispatchStarted: () => { + inFlight += 1; + }, + dispatchReturned: () => { + inFlight -= 1; + lastReturnedAt = Date.now(); + }, + }; +}; + +const shouldInterruptAfterDeadline = + (deadline: DeadlineTracker): (() => boolean) => + () => { + const deadlineMs = deadline.deadlineMs(); + return deadlineMs !== null && Date.now() >= deadlineMs; + }; + +const normalizeExecutionError = ( + cause: unknown, + deadline: DeadlineTracker, + timeoutMs: number, +): string => { const message = toErrorMessage(cause); - return Date.now() >= deadlineMs && looksLikeInterruptedError(message) + const deadlineMs = deadline.deadlineMs(); + return deadlineMs !== null && Date.now() >= deadlineMs && looksLikeInterruptedError(message) ? timeoutMessage(timeoutMs) : message; }; @@ -252,6 +287,7 @@ const createToolBridge = ( toolInvoker: SandboxToolInvoker, pendingDeferreds: Set, runPromise: RunPromise, + deadline: DeadlineTracker, ): QuickJSHandle => context.newFunction("__executor_invokeTool", (pathHandle, argsHandle) => { const path = context.getString(pathHandle); @@ -265,8 +301,10 @@ const createToolBridge = ( pendingDeferreds.delete(deferred); }); + deadline.dispatchStarted(); void runPromise(toolInvoker.invoke({ path, args })).then( (value) => { + deadline.dispatchReturned(); if (!deferred.alive) { return; } @@ -282,6 +320,7 @@ const createToolBridge = ( valueHandle.dispose(); }, (cause) => { + deadline.dispatchReturned(); if (!deferred.alive) { return; } @@ -308,11 +347,12 @@ const createToolBridge = ( const drainJobs = ( context: QuickJSContext, runtime: QuickJSRuntime, - deadlineMs: number, + deadline: DeadlineTracker, timeoutMs: number, ): void => { while (runtime.hasPendingJob()) { - if (Date.now() >= deadlineMs) { + const deadlineMs = deadline.deadlineMs(); + if (deadlineMs !== null && Date.now() >= deadlineMs) { throw new Error(timeoutMessage(timeoutMs)); } @@ -327,9 +367,16 @@ const drainJobs = ( const waitForDeferreds = async ( pendingDeferreds: ReadonlySet, - deadlineMs: number, + deadline: DeadlineTracker, timeoutMs: number, ): Promise => { + const settled = Promise.race([...pendingDeferreds].map((deferred) => deferred.settled)); + const deadlineMs = deadline.deadlineMs(); + if (deadlineMs === null) { + await settled; + return; + } + const remainingMs = deadlineMs - Date.now(); if (remainingMs <= 0) { throw new Error(timeoutMessage(timeoutMs)); @@ -338,7 +385,7 @@ const waitForDeferreds = async ( let timer: ReturnType | undefined; try { await Promise.race([ - Promise.race([...pendingDeferreds].map((deferred) => deferred.settled)), + settled, new Promise((_, reject) => { timer = setTimeout(() => reject(new Error(timeoutMessage(timeoutMs))), remainingMs); }), @@ -354,17 +401,17 @@ const drainAsync = async ( context: QuickJSContext, runtime: QuickJSRuntime, pendingDeferreds: ReadonlySet, - deadlineMs: number, + deadline: DeadlineTracker, timeoutMs: number, ): Promise => { - drainJobs(context, runtime, deadlineMs, timeoutMs); + drainJobs(context, runtime, deadline, timeoutMs); while (pendingDeferreds.size > 0) { - await waitForDeferreds(pendingDeferreds, deadlineMs, timeoutMs); - drainJobs(context, runtime, deadlineMs, timeoutMs); + await waitForDeferreds(pendingDeferreds, deadline, timeoutMs); + drainJobs(context, runtime, deadline, timeoutMs); } - drainJobs(context, runtime, deadlineMs, timeoutMs); + drainJobs(context, runtime, deadline, timeoutMs); }; const evaluateInQuickJs = async ( @@ -374,7 +421,7 @@ const evaluateInQuickJs = async ( runPromise: RunPromise, ): Promise => { const timeoutMs = Math.max(100, options.timeoutMs ?? DEFAULT_TIMEOUT_MS); - const deadlineMs = Date.now() + timeoutMs; + const deadline = makeDeadlineTracker(timeoutMs); const logs: string[] = []; const pendingDeferreds = new Set(); const QuickJS = await resolveQuickJS(); @@ -384,7 +431,7 @@ const evaluateInQuickJs = async ( runtime.setMemoryLimit(options.memoryLimitBytes ?? DEFAULT_MEMORY_LIMIT_BYTES); runtime.setMaxStackSize(options.maxStackSizeBytes ?? DEFAULT_MAX_STACK_SIZE_BYTES); - runtime.setInterruptHandler(shouldInterruptAfterDeadline(deadlineMs)); + runtime.setInterruptHandler(shouldInterruptAfterDeadline(deadline)); const context = runtime.newContext(); try { @@ -392,7 +439,13 @@ const evaluateInQuickJs = async ( context.setProp(context.global, "__executor_log", logBridge); logBridge.dispose(); - const toolBridge = createToolBridge(context, toolInvoker, pendingDeferreds, runPromise); + const toolBridge = createToolBridge( + context, + toolInvoker, + pendingDeferreds, + runPromise, + deadline, + ); context.setProp(context.global, "__executor_invokeTool", toolBridge); toolBridge.dispose(); @@ -402,7 +455,7 @@ const evaluateInQuickJs = async ( evaluated.error.dispose(); return { result: null, - error: normalizeExecutionError(error, deadlineMs, timeoutMs), + error: normalizeExecutionError(error, deadline, timeoutMs), logs, } satisfies ExecuteResult; } @@ -418,14 +471,14 @@ const evaluateInQuickJs = async ( stateResult.error.dispose(); return { result: null, - error: normalizeExecutionError(error, deadlineMs, timeoutMs), + error: normalizeExecutionError(error, deadline, timeoutMs), logs, } satisfies ExecuteResult; } const stateHandle = stateResult.value; try { - await drainAsync(context, runtime, pendingDeferreds, deadlineMs, timeoutMs); + await drainAsync(context, runtime, pendingDeferreds, deadline, timeoutMs); const state = readResultState(context, stateHandle); if (!state.settled) { return { @@ -439,7 +492,7 @@ const evaluateInQuickJs = async ( if (typeof state.error !== "undefined") { return { result: null, - error: normalizeExecutionError(state.error, deadlineMs, timeoutMs), + error: normalizeExecutionError(state.error, deadline, timeoutMs), output: readOutputItems(context), logs, } satisfies ExecuteResult; @@ -466,7 +519,7 @@ const evaluateInQuickJs = async ( } catch (cause) { return { result: null, - error: normalizeExecutionError(cause, deadlineMs, timeoutMs), + error: normalizeExecutionError(cause, deadline, timeoutMs), logs, } satisfies ExecuteResult; } finally {