History

refactor(execution): flip engine to Effect-native, propagate typed errors

## Motivation `McpSessionDO.init` span on the dashboard showed a single 2.77s root with no children. Sub-spans inside the engine (postgres queries, MCP server registration) never appeared as children because every public engine method was a Promise wrapping `Effect.runPromise` internally — each call spawned a new root trace and orphaned the work done underneath. Also: `CodeExecutor.execute` returned `Effect<ExecuteResult, unknown>`, which forced `cause instanceof Error ? cause.message : String(cause)` boilerplate at every error boundary and made Sentry / dashboard filtering impossible to type. ## What changed `@executor/execution`'s `engine.ts` is now Effect-native: - `ExecutionEngine<E extends Cause.YieldableError = CodeExecutionError>` — methods return `Effect<..., E>` not `Promise<...>`. - `createExecutionEngine` propagates the runtime's typed error through. - `InternalPausedExecution<E>` follows. `@executor/execution/promise` becomes the sole home for `Effect.runPromise`: - `createExecutionEngine` (Promise API) unchanged externally — same signatures callers were using. - New `toPromiseExecutionEngine(effectEngine)` lets Effect-aware hosts that need a Promise adapter (e.g. the `ExecutionEngineService` consumers) convert at a single edge. `@executor/codemode-core`'s `CodeExecutor`: - Now `CodeExecutor<E extends Cause.YieldableError = CodeExecutionError>`. - Default error is the new `CodeExecutionError` tagged error. - Every runtime parameterizes with its own tagged error: - `runtime-dynamic-worker` → `CodeExecutor<DynamicWorkerExecutionError>` - `runtime-quickjs` → `CodeExecutor<QuickJsExecutionError>` - `runtime-secure-exec` / `runtime-deno-subprocess` → `CodeExecutor<never>` (both already catch internally and fold errors into `ExecuteResult`). `host-mcp` stays Promise-at-the-boundary — MCP SDK callbacks are async — but now imports from `@executor/execution` (Effect) and calls `Effect.runPromise` only at the SDK callback boundary, where it belongs. Factory is generic in `E` so callers flow their runtime-specific error through unchanged. `@executor/api`'s `ExecutionEngineService` is Effect-typed (`ExecutionEngine<Cause.YieldableError>`); its `handlers/executions.ts` drops every `Effect.promise(() => engine.x(...))` wrapper in favour of `yield* engine.x(...)`. ## Apps/cloud fallout - `services/execution-stack.ts`: returns a single Effect engine; no Promise wrapper path. - `api/execution-usage.ts`: `withExecutionUsageTracking<E>` is Effect- native and generic — `Effect.tap` to record usage, no `async` wrappers, no `Promise<...>` plumbing. - `mcp-session.ts`: `createConnectedRuntimeEffect` now yields `buildExecuteDescription(executor)` directly, so the two postgres queries inside (`executor.sources.list`, `executor.tools.list`) land as children of `McpSessionDO.createRuntime` in the trace. - `api/protected.ts` + test harnesses: pass Effect engines to `ExecutionEngineService` directly. ## Verified in prod Init trace tree after deploy (sample trace `16b2e150e986d703b3b23adad6ae3816`): ``` McpSessionDO.init 2.51s ├── resolveSessionMeta 231ms └── createRuntime 2.28s ├── makeExecutionStack → createScopedExecutor 0ms ├── buildExecuteDescription 2.28s │ ├── executor.sources.list 167ms │ └── executor.tools.list 2.11s ← real bottleneck ├── createExecutorMcpServer 0ms └── transport.connect 0ms ``` `executor.tools.list` is now visible as the 2.1s postgres scan — the actual optimization target. Before this refactor it was hidden under an opaque 2.77s `init` span.

Rhys Sullivan committed 6 months ago

Commit 2fa41671 parent ce44a5918 files · +256 −191Browse files

Unified · Split

18 files
  1. apps/cloud/src/api/execution-usage.ts +16 −18
  2. apps/cloud/src/api/protected.test.ts +22 −26
  3. apps/cloud/src/api/protected.ts +0 −3
  4. apps/cloud/src/mcp-session.ts +9 −2
  5. packages/core/api/src/handlers/executions.ts +5 −7
  6. packages/core/api/src/services.ts +7 −1
  7. packages/core/execution/src/description.ts +7 −3
  8. packages/core/execution/src/engine.ts +35 −30
  9. packages/core/execution/src/errors.ts +5 −0
  10. packages/core/execution/src/promise.ts +45 −30
  11. packages/core/execution/src/tool-invoker.test.ts +39 −51
  12. packages/hosts/mcp/src/server.ts +28 −13
  13. packages/kernel/core/src/effect-errors.ts +14 −0
  14. packages/kernel/core/src/types.ts +14 −3
  15. packages/kernel/runtime-deno-subprocess/src/index.ts +1 −1
  16. packages/kernel/runtime-dynamic-worker/src/executor.ts +3 −1
  17. packages/kernel/runtime-quickjs/src/index.ts +3 −1
  18. packages/kernel/runtime-secure-exec/src/index.ts +3 −1
apps/cloud/src/api/execution-usage.ts+16−18
@@ -1,24 +1,22 @@
1+import { Effect } from "effect";
2+import type * as Cause from "effect/Cause";
3+
14 import type { ExecutionEngine } from "@executor/execution";
25
3−export const withExecutionUsageTracking = (
6+export const withExecutionUsageTracking = <E extends Cause.YieldableError>(
47 organizationId: string,
5− engine: ExecutionEngine,
8+ engine: ExecutionEngine<E>,
69 trackUsage: (organizationId: string) => void,
7−): ExecutionEngine => ({
8− execute: async (code, options) => {
9− const result = await engine.execute(code, options);
10− trackUsage(organizationId);
11− return result;
12− },
13− executeWithPause: async (code) => {
14− const result = await engine.executeWithPause(code);
15− trackUsage(organizationId);
16− return result;
17− },
18− resume: async (executionId, response) => {
19− const result = await engine.resume(executionId, response);
20− // resume doesn't count as usage
21− return result;
22− },
10+): ExecutionEngine<E> => ({
11+ execute: (code, options) =>
12+ engine
13+ .execute(code, options)
14+ .pipe(Effect.tap(() => Effect.sync(() => trackUsage(organizationId)))),
15+ executeWithPause: (code) =>
16+ engine
17+ .executeWithPause(code)
18+ .pipe(Effect.tap(() => Effect.sync(() => trackUsage(organizationId)))),
19+ // resume doesn't count as usage
20+ resume: (executionId, response) => engine.resume(executionId, response),
2321 getDescription: engine.getDescription,
2422 });
apps/cloud/src/api/protected.test.ts+22−26
@@ -5,16 +5,18 @@ import { withExecutionUsageTracking } from "./execution-usage"; · 4 unmodified lines
55
66 const makeBaseEngine = (): ExecutionEngine =>
77 ({
8− execute: async () => ({ result: "ok", logs: [] }),
9− executeWithPause: async () => ({
10− status: "completed",
11− result: { result: "ok", logs: [] },
12− }),
13− resume: async () => ({
14− status: "completed",
15− result: { result: "ok", logs: [] },
16− }),
17− getDescription: async () => "desc",
8+ execute: () => Effect.succeed({ result: "ok", logs: [] }),
9+ executeWithPause: () =>
10+ Effect.succeed({
11+ status: "completed",
12+ result: { result: "ok", logs: [] },
13+ }),
14+ resume: () =>
15+ Effect.succeed({
16+ status: "completed",
17+ result: { result: "ok", logs: [] },
18+ }),
19+ getDescription: Effect.succeed("desc"),
1820 }) as ExecutionEngine;
1921
2022 describe("withExecutionUsageTracking", () => {
@@ -25,10 +27,8 @@ describe("withExecutionUsageTracking", () => { · 4 unmodified lines
2527 tracked.push(orgId);
2628 });
2729
28− yield* Effect.promise(() =>
29− engine.execute("1+1", { onElicitation: (() => Effect.die("unused")) as never }),
30− );
31− yield* Effect.promise(() => engine.executeWithPause("2+2"));
30+ yield* engine.execute("1+1", { onElicitation: (() => Effect.die("unused")) as never });
31+ yield* engine.executeWithPause("2+2");
3232
3333 expect(tracked).toEqual(["org_1", "org_1"]);
3434 }),
@@ -44,8 +44,8 @@ describe("withExecutionUsageTracking", () => { · 9 unmodified lines
4444 "org_2",
4545 {
4646 ...base,
47− resume: async (...args) => {
48− if (shouldReturnNull) return null;
47+ resume: (...args) => {
48+ if (shouldReturnNull) return Effect.succeed(null);
4949 return base.resume(...args);
5050 },
5151 },
@@ -54,17 +54,13 @@ describe("withExecutionUsageTracking", () => { · 2 unmodified lines
5454 },
5555 );
5656
57− yield* Effect.promise(() =>
58− engine.resume("exec_1", {
59− action: "accept",
60− }),
61− );
57+ yield* engine.resume("exec_1", {
58+ action: "accept",
59+ });
6260 shouldReturnNull = true;
63− yield* Effect.promise(() =>
64− engine.resume("missing", {
65− action: "accept",
66− }),
67− );
61+ yield* engine.resume("missing", {
62+ action: "accept",
63+ });
6864
6965 expect(tracked).toEqual([]);
7066 }),
apps/cloud/src/api/protected.ts+0−3
@@ -34,9 +34,6 @@ const createProtectedApp = (organizationId: string, organizationName: string) => · 33 unmodified lines
3434 Effect.gen(function* () {
3535 const { executor, engine } = yield* makeExecutionStack(organizationId, organizationName);
3636
37− // Handlers wrap their own bodies with `capture(...)` — the edge
38− // translation lives per-handler, not at service construction. See
39− // notes/error-handling.md.
4037 const requestServices = Layer.mergeAll(
4138 Layer.succeed(ExecutorService, executor),
4239 Layer.succeed(ExecutionEngineService, engine),
apps/cloud/src/mcp-session.ts+9−2
@@ -11,6 +11,7 @@ import { drizzle } from "drizzle-orm/postgres-js"; · 10 unmodified lines
1111 import postgres from "postgres";
1212
1313 import { createExecutorMcpServer } from "@executor/host-mcp";
14+import { buildExecuteDescription } from "@executor/execution";
1415 import type { DrizzleDb, DbServiceShape } from "./services/db";
1516
1617 // Import directly from core-shared-services, NOT from ./api/layers.ts.
@@ -176,12 +177,18 @@ export class McpSessionDO extends DurableObject { · 159 unmodified lines
176177 ) {
177178 const self = this;
178179 return Effect.gen(function* () {
179− const { engine } = yield* makeExecutionStack(
180+ const { executor, engine } = yield* makeExecutionStack(
180181 sessionMeta.organizationId,
181182 sessionMeta.organizationName,
182183 );
184+ // Build the description here so the two postgres queries it runs
185+ // (`executor.sources.list` + `executor.tools.list`) land as
186+ // children of `McpSessionDO.createRuntime`. host-mcp would
187+ // otherwise call `Effect.runPromise(engine.getDescription)` at
188+ // its async MCP-SDK boundary and orphan those sub-spans.
189+ const description = yield* buildExecuteDescription(executor);
183190 const mcpServer = yield* Effect.promise(() =>
184− createExecutorMcpServer({ engine }),
191+ createExecutorMcpServer({ engine, description }),
185192 ).pipe(Effect.withSpan("McpSessionDO.createExecutorMcpServer"));
186193 const transport = new WorkerTransport({
187194 sessionIdGenerator: () => self.ctx.id.toString(),
packages/core/api/src/handlers/executions.ts+5−7
@@ -11,7 +11,7 @@ export const ExecutionsHandlers = HttpApiBuilder.group(ExecutorApi, "executions" · 10 unmodified lines
1111 .handle("execute", ({ payload }) =>
1212 capture(Effect.gen(function* () {
1313 const engine = yield* ExecutionEngineService;
14− const outcome = yield* Effect.promise(() => engine.executeWithPause(payload.code));
14+ const outcome = yield* engine.executeWithPause(payload.code);
1515
1616 if (outcome.status === "completed") {
1717 const formatted = formatExecuteResult(outcome.result);
@@ -34,12 +34,10 @@ export const ExecutionsHandlers = HttpApiBuilder.group(ExecutorApi, "executions" · 16 unmodified lines
3434 .handle("resume", ({ path, payload }) =>
3535 capture(Effect.gen(function* () {
3636 const engine = yield* ExecutionEngineService;
37− const result = yield* Effect.promise(() =>
38− engine.resume(path.executionId, {
39− action: payload.action,
40− content: payload.content as Record<string, unknown> | undefined,
41− }),
42− );
37+ const result = yield* engine.resume(path.executionId, {
38+ action: payload.action,
39+ content: payload.content as Record<string, unknown> | undefined,
40+ });
4341
4442 if (!result) {
4543 return yield* Effect.fail({
packages/core/api/src/services.ts+7−1
@@ -1,4 +1,5 @@
11 import { Context } from "effect";
2+import type * as Cause from "effect/Cause";
23 import type { Executor } from "@executor/sdk";
34 import type { ExecutionEngine } from "@executor/execution";
45
@@ -7,7 +8,12 @@ export class ExecutorService extends Context.Tag("ExecutorService")< · 2 unmodified lines
78 Executor
89 >() {}
910
11+// Error channel widened to `Cause.YieldableError` so callers that plug
12+// in a runtime-specific tagged error (e.g.
13+// `ExecutionEngine<DynamicWorkerExecutionError>`) assign structurally.
14+// Handlers yield directly; defects flow through `Effect.catchAllCause`
15+// at the edge.
1016 export class ExecutionEngineService extends Context.Tag("ExecutionEngineService")<
1117 ExecutionEngineService,
12− ExecutionEngine
18+ ExecutionEngine<Cause.YieldableError>
1319 >() {}
packages/core/execution/src/description.ts+7−3
@@ -10,14 +10,18 @@ import type { Executor, Tool, Source } from "@executor/sdk"; · 9 unmodified lines
1010 */
1111 export const buildExecuteDescription = (executor: Executor): Effect.Effect<string> =>
1212 Effect.gen(function* () {
13− const sources: readonly Source[] = yield* executor.sources.list().pipe(Effect.orDie);
14− const tools: readonly Tool[] = yield* executor.tools.list().pipe(Effect.orDie);
13+ const sources: readonly Source[] = yield* executor.sources
14+ .list()
15+ .pipe(Effect.orDie, Effect.withSpan("executor.sources.list"));
16+ const tools: readonly Tool[] = yield* executor.tools
17+ .list()
18+ .pipe(Effect.orDie, Effect.withSpan("executor.tools.list"));
1519
1620 const namespaces = new Set<string>();
1721 for (const tool of tools) namespaces.add(tool.sourceId);
1822
1923 return formatDescription([...namespaces], sources);
20− });
24+ }).pipe(Effect.withSpan("buildExecuteDescription"));
2125
2226 const formatDescription = (namespaces: readonly string[], sources: readonly Source[]): string => {
2327 const lines: string[] = [
packages/core/execution/src/engine.ts+35−30
@@ -1,4 +1,5 @@
11 import { Deferred, Effect, Fiber, Ref } from "effect";
2+import type * as Cause from "effect/Cause";
23
34 import type {
45 Executor,
@@ -7,6 +8,7 @@ import type { · 2 unmodified lines
78 ElicitationHandler,
89 ElicitationContext,
910 } from "@executor/sdk";
11+import { CodeExecutionError } from "@executor/codemode-core";
1012 import type { CodeExecutor, ExecuteResult, SandboxToolInvoker } from "@executor/codemode-core";
1113
1214 import {
@@ -22,9 +24,11 @@ import { buildExecuteDescription } from "./description"; · 9 unmodified lines
2224 // Types
2325 // ---------------------------------------------------------------------------
2426
25−export type ExecutionEngineConfig = {
27+export type ExecutionEngineConfig<
28+ E extends Cause.YieldableError = CodeExecutionError,
29+> = {
2630 readonly executor: Executor;
27− readonly codeExecutor: CodeExecutor;
31+ readonly codeExecutor: CodeExecutor<E>;
2832 };
2933
3034 export type ExecutionResult =
@@ -37,10 +41,10 @@ export type PausedExecution = { · 6 unmodified lines
3741 };
3842
3943 /** Internal representation with Effect runtime state for pause/resume. */
40−type InternalPausedExecution = PausedExecution & {
44+type InternalPausedExecution<E> = PausedExecution & {
4145 readonly response: Deferred.Deferred<typeof ElicitationResponse.Type>;
42− readonly fiber: Fiber.Fiber<ExecuteResult, unknown>;
43− readonly pauseSignalRef: Ref.Ref<Deferred.Deferred<InternalPausedExecution>>;
46+ readonly fiber: Fiber.Fiber<ExecuteResult, E>;
47+ readonly pauseSignalRef: Ref.Ref<Deferred.Deferred<InternalPausedExecution<E>>>;
4448 };
4549
4650 export type ResumeResponse = {
@@ -255,22 +259,26 @@ const makeFullInvoker = (executor: Executor, invokeOptions: InvokeOptions): Sand · 208 unmodified lines
255259 // Execution Engine
256260 // ---------------------------------------------------------------------------
257261
258−export type ExecutionEngine = {
262+export type ExecutionEngine<E extends Cause.YieldableError = CodeExecutionError> = {
259263 /**
260264 * Execute code with elicitation handled inline by the provided handler.
261265 * Use this when the host supports elicitation (e.g. MCP with elicitation capability).
266+ *
267+ * Fails with the code executor's typed error `E` (defaults to
268+ * `CodeExecutionError`). Runtimes surface their own `Data.TaggedError`
269+ * subclass, which flows through here unchanged.
262270 */
263271 readonly execute: (
264272 code: string,
265273 options: { readonly onElicitation: ElicitationHandler },
266− ) => Promise<ExecuteResult>;
274+ ) => Effect.Effect<ExecuteResult, E>;
267275
268276 /**
269277 * Execute code, intercepting the first elicitation as a pause point.
270278 * Use this when the host doesn't support inline elicitation.
271279 * Returns either a completed result or a paused execution that can be resumed.
272280 */
273− readonly executeWithPause: (code: string) => Promise<ExecutionResult>;
281+ readonly executeWithPause: (code: string) => Effect.Effect<ExecutionResult, E>;
274282
275283 /**
276284 * Resume a paused execution. Returns a completed result, a new pause, or
@@ -279,20 +287,21 @@ export type ExecutionEngine = { · 2 unmodified lines
279287 readonly resume: (
280288 executionId: string,
281289 response: ResumeResponse,
282− ) => Promise<ExecutionResult | null>;
290+ ) => Effect.Effect<ExecutionResult | null, E>;
283291
284292 /**
285293 * Get the dynamic tool description (workflow + namespaces).
286294 */
287− readonly getDescription: () => Promise<string>;
295+ readonly getDescription: Effect.Effect<string>;
288296 };
289297
290−const runEffect = <A>(effect: Effect.Effect<A, unknown>): Promise<A> =>
291− Effect.runPromise(effect as Effect.Effect<A, never>);
292−
293−export const createExecutionEngine = (config: ExecutionEngineConfig): ExecutionEngine => {
298+export const createExecutionEngine = <
299+ E extends Cause.YieldableError = CodeExecutionError,
300+>(
301+ config: ExecutionEngineConfig<E>,
302+): ExecutionEngine<E> => {
294303 const { executor, codeExecutor } = config;
295− const pausedExecutions = new Map<string, InternalPausedExecution>();
304+ const pausedExecutions = new Map<string, InternalPausedExecution<E>>();
296305 let nextId = 0;
297306
298307 /**
@@ -301,12 +310,11 @@ export const createExecutionEngine = (config: ExecutionEngineConfig): ExecutionE · 2 unmodified lines
301310 * comes first). Re-used by both executeWithPause and resume.
302311 */
303312 const awaitCompletionOrPause = (
304− fiber: Fiber.Fiber<ExecuteResult, unknown>,
305− pauseSignal: Deferred.Deferred<InternalPausedExecution>,
306− ): Effect.Effect<ExecutionResult> =>
313+ fiber: Fiber.Fiber<ExecuteResult, E>,
314+ pauseSignal: Deferred.Deferred<InternalPausedExecution<E>>,
315+ ): Effect.Effect<ExecutionResult, E> =>
307316 Effect.race(
308317 Fiber.join(fiber).pipe(
309− Effect.orDie,
310318 Effect.map((result): ExecutionResult => ({ status: "completed", result })),
311319 ),
312320 Deferred.await(pauseSignal).pipe(
@@ -329,17 +337,17 @@ export const createExecutionEngine = (config: ExecutionEngineConfig): ExecutionE · 16 unmodified lines
329337 // Ref holds the current pause signal. The elicitation handler reads
330338 // it each time it fires, so resume() can swap in a fresh Deferred
331339 // before unblocking the fiber.
332− const pauseSignalRef = yield* Ref.make(yield* Deferred.make<InternalPausedExecution>());
340+ const pauseSignalRef = yield* Ref.make(yield* Deferred.make<InternalPausedExecution<E>>());
333341
334342 // Will be set once the fiber is forked.
335− let fiber: Fiber.Fiber<ExecuteResult, unknown>;
343+ let fiber: Fiber.Fiber<ExecuteResult, E>;
336344
337345 const elicitationHandler: ElicitationHandler = (ctx) =>
338346 Effect.gen(function* () {
339347 const responseDeferred = yield* Deferred.make<typeof ElicitationResponse.Type>();
340348 const id = `exec_${++nextId}`;
341349
342− const paused: InternalPausedExecution = {
350+ const paused: InternalPausedExecution<E> = {
343351 id,
344352 elicitationContext: ctx,
345353 response: responseDeferred,
@@ -383,7 +391,7 @@ export const createExecutionEngine = (config: ExecutionEngineConfig): ExecutionE · 37 unmodified lines
383391
384392 // Swap in a fresh pause signal BEFORE unblocking the fiber, so the
385393 // next elicitation handler call signals this new Deferred.
386− const nextSignal = yield* Deferred.make<InternalPausedExecution>();
394+ const nextSignal = yield* Deferred.make<InternalPausedExecution<E>>();
387395 yield* Ref.set(paused.pauseSignalRef, nextSignal);
388396
389397 yield* Deferred.succeed(paused.response, {
@@ -415,12 +423,9 @@ export const createExecutionEngine = (config: ExecutionEngineConfig): ExecutionE · 25 unmodified lines
415423 });
416424
417425 return {
418− execute: (code, options) => runEffect(runInlineExecution(code, options)),
419−
420− executeWithPause: (code) => runEffect(startPausableExecution(code)),
421−
422− resume: (executionId, response) => runEffect(resumeExecution(executionId, response)),
423−
424− getDescription: () => runEffect(buildExecuteDescription(executor)),
426+ execute: runInlineExecution,
427+ executeWithPause: startPausableExecution,
428+ resume: resumeExecution,
429+ getDescription: buildExecuteDescription(executor),
425430 };
426431 };
packages/core/execution/src/errors.ts+5−0
@@ -4,3 +4,8 @@ export class ExecutionToolError extends Data.TaggedError("ExecutionToolError")<{ · 3 unmodified lines
44 readonly message: string;
55 readonly cause?: unknown;
66 }> {}
7+
8+// `CodeExecutionError` lives in `@executor/codemode-core` — the `CodeExecutor`
9+// interface uses it as the default error channel, so the runtime packages
10+// can import the same class directly.
11+export { CodeExecutionError } from "@executor/codemode-core";
packages/core/execution/src/promise.ts+45−30
@@ -1,24 +1,28 @@
11 // ---------------------------------------------------------------------------
22 // @executor/execution/promise — Promise-native surface for the execution
3−// engine. Accepts a Promise-style Executor (from @executor/sdk) and an
4−// async `onElicitation` handler — no Effect imports required by callers.
3+// engine.
4+// ---------------------------------------------------------------------------
55 //
6−// Under the hood the engine is Effect-based, so we wrap the incoming
7−// Promise executor into the minimal Effect shape the engine consumes,
8−// and bridge the elicitation handler via `Effect.tryPromise`.
6+// `engine.ts` is Effect-native; this module runs each method with
7+// `Effect.runPromise` at the boundary so hosts that can't compose Effects
8+// (the MCP SDK tool handlers, plain async call sites) can still use the
9+// engine. Callers already inside an Effect context should import directly
10+// from `@executor/execution` to keep trace context intact.
911 // ---------------------------------------------------------------------------
1012
1113 import { Effect } from "effect";
14+import type * as Cause from "effect/Cause";
1215
1316 import type {
1417 ElicitationContext,
1518 ElicitationResponse,
1619 Executor as PromiseExecutor,
1720 } from "@executor/sdk";
18−import type { CodeExecutor, ExecuteResult } from "@executor/codemode-core";
21+import type { CodeExecutionError, CodeExecutor, ExecuteResult } from "@executor/codemode-core";
1922
2023 import {
2124 createExecutionEngine as createEffectExecutionEngine,
25+ type ExecutionEngine as EffectExecutionEngine,
2226 type ExecutionResult,
2327 type PausedExecution,
2428 type ResumeResponse,
@@ -28,9 +32,11 @@ export type ElicitationHandler = ( · 3 unmodified lines
2832 ctx: ElicitationContext,
2933 ) => Promise<ElicitationResponse>;
3034
31−export type ExecutionEngineConfig = {
35+export type ExecutionEngineConfig<
36+ E extends Cause.YieldableError = CodeExecutionError,
37+> = {
3238 readonly executor: PromiseExecutor;
33− readonly codeExecutor: CodeExecutor;
39+ readonly codeExecutor: CodeExecutor<E>;
3440 };
3541
3642 export type ExecutionEngine = {
@@ -49,7 +55,7 @@ export type ExecutionEngine = { · 12 unmodified lines
4955 /**
5056 * Wrap a Promise-style executor into the Effect shape the engine consumes.
5157 * Only the four method families the engine actually touches need wrapping:
52− * `tools.{invoke,list,schema}` and `sources.list`.
58+ * `tools.{invoke,list,schema,definitions}` and `sources.list`.
5359 */
5460 const wrapPromiseExecutor = (pe: PromiseExecutor): any => ({
5561 scope: (pe as any).scope,
@@ -84,35 +90,44 @@ const wrapPromiseExecutor = (pe: PromiseExecutor): any => ({ · 28 unmodified lines
8490 },
8591 });
8692
87−export const createExecutionEngine = (
88− config: ExecutionEngineConfig,
89−): ExecutionEngine => {
90− const engine = createEffectExecutionEngine({
91− executor: wrapPromiseExecutor(config.executor),
92− codeExecutor: config.codeExecutor,
93− });
94− return {
95− execute: (code, options) =>
93+/**
94+ * Promise-wrap an Effect-native `ExecutionEngine` (from `./engine`).
95+ * Exposed separately so callers that already hold an Effect engine
96+ * (apps/cloud's execution-stack composes both) can convert it for hosts
97+ * that need the Promise surface (host-mcp).
98+ */
99+export const toPromiseExecutionEngine = <E extends Cause.YieldableError>(
100+ engine: EffectExecutionEngine<E>,
101+): ExecutionEngine => ({
102+ execute: (code, options) =>
103+ Effect.runPromise(
96104 engine.execute(code, {
97105 onElicitation: (ctx) =>
98− Effect.tryPromise(() => options.onElicitation(ctx)).pipe(
99− Effect.orDie,
100− ),
106+ Effect.tryPromise(() => options.onElicitation(ctx)).pipe(Effect.orDie),
101107 }),
102− executeWithPause: (code) => engine.executeWithPause(code),
103− resume: (executionId, response) => engine.resume(executionId, response),
104− getDescription: () => engine.getDescription(),
105− };
106−};
108+ ),
109+ executeWithPause: (code) => Effect.runPromise(engine.executeWithPause(code)),
110+ resume: (executionId, response) => Effect.runPromise(engine.resume(executionId, response)),
111+ getDescription: () => Effect.runPromise(engine.getDescription),
112+});
113+
114+export const createExecutionEngine = <
115+ E extends Cause.YieldableError = CodeExecutionError,
116+>(
117+ config: ExecutionEngineConfig<E>,
118+): ExecutionEngine =>
119+ toPromiseExecutionEngine(
120+ createEffectExecutionEngine({
121+ executor: wrapPromiseExecutor(config.executor),
122+ codeExecutor: config.codeExecutor,
123+ }),
124+ );
107125
108126 // ---------------------------------------------------------------------------
109127 // Re-exports — plain types/helpers that don't carry Effect signatures.
110128 // ---------------------------------------------------------------------------
111129
112−export {
113− formatExecuteResult,
114− formatPausedExecution,
115−} from "./engine";
130+export { formatExecuteResult, formatPausedExecution } from "./engine";
116131
117132 export type { ExecutionResult, PausedExecution, ResumeResponse };
118133
packages/core/execution/src/tool-invoker.test.ts+39−51
@@ -150,11 +150,9 @@ describe("tool discovery", () => { · 149 unmodified lines
150150 });
151151 expect(crmOnly.map((match) => match.path)).toEqual(["crm.listContacts"]);
152152
153− const sandboxResult = yield* Effect.promise(() =>
154− createExecutionEngine({ executor, codeExecutor }).execute(
155− 'return await tools.search({ namespace: "crm", query: "create contact", limit: 5 });',
156− { onElicitation: acceptAll },
157− ),
153+ const sandboxResult = yield* createExecutionEngine({ executor, codeExecutor }).execute(
154+ 'return await tools.search({ namespace: "crm", query: "create contact", limit: 5 });',
155+ { onElicitation: acceptAll },
158156 );
159157 expect(sandboxResult.error).toBeUndefined();
160158 expect(sandboxResult.result).toEqual([
@@ -167,10 +165,9 @@ describe("tool discovery", () => { · 6 unmodified lines
167165 Effect.gen(function* () {
168166 const executor = yield* makeSearchExecutor();
169167
170− const listed = yield* Effect.promise(() =>
171− createExecutionEngine({ executor, codeExecutor }).execute("return await tools.executor.sources.list();", {
172− onElicitation: acceptAll,
173− }),
168+ const listed = yield* createExecutionEngine({ executor, codeExecutor }).execute(
169+ "return await tools.executor.sources.list();",
170+ { onElicitation: acceptAll },
174171 );
175172 expect(listed.error).toBeUndefined();
176173 expect(listed.result).toEqual(
@@ -180,11 +177,9 @@ describe("tool discovery", () => { · 3 unmodified lines
180177 ]),
181178 );
182179
183− const searched = yield* Effect.promise(() =>
184− createExecutionEngine({ executor, codeExecutor }).execute(
185− 'return await tools.search({ query: "list contacts", namespace: "crm", limit: 5 });',
186− { onElicitation: acceptAll },
187− ),
180+ const searched = yield* createExecutionEngine({ executor, codeExecutor }).execute(
181+ 'return await tools.search({ query: "list contacts", namespace: "crm", limit: 5 });',
182+ { onElicitation: acceptAll },
188183 );
189184 expect(searched.error).toBeUndefined();
190185 expect(searched.result).toEqual([expect.objectContaining({ path: "crm.listContacts" })]);
@@ -210,55 +205,48 @@ describe("tool discovery", () => { · 19 unmodified lines
210205 const executor = yield* makeSearchExecutor();
211206 const engine = createExecutionEngine({ executor, codeExecutor });
212207
213− const invalid = yield* Effect.promise(() =>
214− engine.execute(
215− [
216− "try {",
217− ' await tools.search("github issues");',
218− ' return "unexpected";',
219− "} catch (error) {",
220− " return error instanceof Error ? error.message : String(error);",
221− "}",
222− ].join("\n"),
223− { onElicitation: acceptAll },
224− ),
208+ const invalid = yield* engine.execute(
209+ [
210+ "try {",
211+ ' await tools.search("github issues");',
212+ ' return "unexpected";',
213+ "} catch (error) {",
214+ " return error instanceof Error ? error.message : String(error);",
215+ "}",
216+ ].join("\n"),
217+ { onElicitation: acceptAll },
225218 );
226219 expect(invalid.error).toBeUndefined();
227220 expect(String(invalid.result)).toContain(
228221 "tools.search expects an object: { query?: string; namespace?: string; limit?: number }",
229222 );
230223
231− const emptyQuery = yield* Effect.promise(() =>
232− engine.execute('return await tools.search({ query: "", limit: 5 });', {
233− onElicitation: acceptAll,
234− }),
224+ const emptyQuery = yield* engine.execute(
225+ 'return await tools.search({ query: "", limit: 5 });',
226+ { onElicitation: acceptAll },
235227 );
236228 expect(emptyQuery.error).toBeUndefined();
237229 expect(emptyQuery.result).toEqual([]);
238230
239− const invalidDescribe = yield* Effect.promise(() =>
240− engine.execute(
241− [
242− "try {",
243− ' await tools.describe.tool({ path: "github.listRepositoryIssues", includeSchemas: true });',
244− ' return "unexpected";',
245− "} catch (error) {",
246− " return error instanceof Error ? error.message : String(error);",
247− "}",
248− ].join("\n"),
249− { onElicitation: acceptAll },
250− ),
231+ const invalidDescribe = yield* engine.execute(
232+ [
233+ "try {",
234+ ' await tools.describe.tool({ path: "github.listRepositoryIssues", includeSchemas: true });',
235+ ' return "unexpected";',
236+ "} catch (error) {",
237+ " return error instanceof Error ? error.message : String(error);",
238+ "}",
239+ ].join("\n"),
240+ { onElicitation: acceptAll },
251241 );
252242 expect(invalidDescribe.error).toBeUndefined();
253243 expect(String(invalidDescribe.result)).toContain(
254244 "tools.describe.tool no longer accepts includeSchemas",
255245 );
256246
257− const invalidSearch = yield* Effect.promise(() =>
258− engine.execute(
259− 'try { return await tools.search("crm"); } catch (error) { return error instanceof Error ? error.message : String(error); }',
260− { onElicitation: acceptAll },
261− ),
247+ const invalidSearch = yield* engine.execute(
248+ 'try { return await tools.search("crm"); } catch (error) { return error instanceof Error ? error.message : String(error); }',
249+ { onElicitation: acceptAll },
262250 );
263251 expect(invalidSearch.error).toBeUndefined();
264252 expect(String(invalidSearch.result)).toContain("tools.search expects an object");
@@ -334,7 +322,7 @@ describe("pause/resume with multiple elicitations", () => { · 69 unmodified lines
334322
335323 const code = "return await tools.api.multiApproval({});";
336324
337− const outcome1 = yield* Effect.promise(() => engine.executeWithPause(code));
325+ const outcome1 = yield* engine.executeWithPause(code);
338326 expect(outcome1.status).toBe("paused");
339327 const paused1 = outcome1 as Extract<typeof outcome1, { status: "paused" }>;
340328 expect(paused1.execution.elicitationContext.request.message).toBe("First approval");
@@ -344,7 +332,7 @@ describe("pause/resume with multiple elicitations", () => { · 3 unmodified lines
344332 // result or the completion).
345333 const outcome2 = yield* Effect.promise(() =>
346334 Promise.race([
347− engine.resume(paused1.execution.id, { action: "accept" }),
335+ Effect.runPromise(engine.resume(paused1.execution.id, { action: "accept" })),
348336 new Promise<never>((_, reject) =>
349337 setTimeout(
350338 () => reject(new Error("resume hung — second elicitation not surfaced")),
@@ -368,7 +356,7 @@ describe("pause/resume with multiple elicitations", () => { · 17 unmodified lines
368356
369357 const code = "return await tools.api.singleApproval({});";
370358
371− const outcome1 = await engine.executeWithPause(code);
359+ const outcome1 = await Effect.runPromise(engine.executeWithPause(code));
372360 expect(outcome1.status).toBe("paused");
373361 const paused1 = outcome1 as Extract<typeof outcome1, { status: "paused" }>;
374362 expect(paused1.execution.elicitationContext.request.message).toBe("Only approval");
@@ -389,7 +377,7 @@ describe("pause/resume with multiple elicitations", () => { · 14 unmodified lines
389377 expect(exitProbe).toBe("still-running");
390378
391379 const outcome2 = await Promise.race([
392− engine.resume(paused1.execution.id, { action: "accept" }),
380+ Effect.runPromise(engine.resume(paused1.execution.id, { action: "accept" })),
393381 new Promise<never>((_, reject) =>
394382 setTimeout(() => reject(new Error("resume hung across runPromise boundaries")), 2000),
395383 ),
packages/hosts/mcp/src/server.ts+28−13
@@ -14,6 +14,7 @@ import type { · 13 unmodified lines
1414 ElicitationContext,
1515 ElicitationRequest,
1616 } from "@executor/sdk";
17+import type * as Cause from "effect/Cause";
1718 import {
1819 createExecutionEngine,
1920 formatExecuteResult,
@@ -44,11 +45,23 @@ class CfWorkerJsonSchemaValidator implements jsonSchemaValidator { · 24 unmodified lines
4445 // Config
4546 // ---------------------------------------------------------------------------
4647
47−export type ExecutorMcpServerConfig =
48− | ExecutionEngineConfig
49− | { readonly engine: ExecutionEngine }
50− | (ExecutionEngineConfig & { readonly stateless: true })
51− | { readonly engine: ExecutionEngine; readonly stateless: true };
48+type SharedMcpServerConfig = {
49+ /**
50+ * Pre-built `execute` tool description. When provided, the factory skips
51+ * its internal `engine.getDescription()` call. Useful when the caller
52+ * wants to compute the description inside its own Effect tracer context
53+ * so sub-spans (`executor.sources.list`, `executor.tools.list`) nest as
54+ * children of the caller's root span instead of being orphaned by the
55+ * `Effect.runPromise` that `engine.getDescription()` runs internally.
56+ */
57+ readonly description?: string;
58+};
59+
60+export type ExecutorMcpServerConfig<E extends Cause.YieldableError = Cause.YieldableError> =
61+ | (ExecutionEngineConfig<E> & SharedMcpServerConfig)
62+ | ({ readonly engine: ExecutionEngine<E> } & SharedMcpServerConfig)
63+ | (ExecutionEngineConfig<E> & SharedMcpServerConfig & { readonly stateless: true })
64+ | ({ readonly engine: ExecutionEngine<E>; readonly stateless: true } & SharedMcpServerConfig);
5265
5366 // ---------------------------------------------------------------------------
5467 // Elicitation bridge
@@ -153,11 +166,11 @@ const toMcpPausedResult = (formatted: ReturnType<typeof formatPausedExecution>): · 98 unmodified lines
153166 // Server factory
154167 // ---------------------------------------------------------------------------
155168
156−export const createExecutorMcpServer = async (
157− config: ExecutorMcpServerConfig,
169+export const createExecutorMcpServer = async <E extends Cause.YieldableError>(
170+ config: ExecutorMcpServerConfig<E>,
158171 ): Promise<McpServer> => {
159172 const engine = "engine" in config ? config.engine : createExecutionEngine(config);
160− const description = await engine.getDescription();
173+ const description = config.description ?? (await Effect.runPromise(engine.getDescription));
161174
162175 const server = new McpServer(
163176 { name: "executor", version: "1.0.0" },
@@ -166,13 +179,15 @@ export const createExecutorMcpServer = async ( · 2 unmodified lines
166179
167180 const executeCode = async (code: string): Promise<McpToolResult> => {
168181 if (supportsManagedElicitation(server)) {
169− const result = await engine.execute(code, {
170− onElicitation: makeMcpElicitationHandler(server),
171− });
182+ const result = await Effect.runPromise(
183+ engine.execute(code, {
184+ onElicitation: makeMcpElicitationHandler(server),
185+ }),
186+ );
172187 return toMcpResult(formatExecuteResult(result));
173188 }
174189
175− const outcome = await engine.executeWithPause(code);
190+ const outcome = await Effect.runPromise(engine.executeWithPause(code));
176191 return outcome.status === "completed"
177192 ? toMcpResult(formatExecuteResult(outcome.result))
178193 : toMcpPausedResult(formatPausedExecution(outcome.execution));
@@ -222,7 +237,7 @@ export const createExecutorMcpServer = async ( · 43 unmodified lines
222237 },
223238 async ({ executionId, action, content: rawContent }) => {
224239 const content = parseJsonContent(rawContent);
225− const outcome = await engine.resume(executionId, { action, content });
240+ const outcome = await Effect.runPromise(engine.resume(executionId, { action, content }));
226241
227242 if (!outcome) {
228243 return {
packages/kernel/core/src/effect-errors.ts+14−0
@@ -7,3 +7,17 @@ export class KernelCoreEffectError extends Data.TaggedError("KernelCoreEffectErr · 6 unmodified lines
77
88 export const kernelCoreEffectError = (module: string, message: string) =>
99 new KernelCoreEffectError({ module, message });
10+
11+/**
12+ * Default failure type for any `CodeExecutor.execute` implementation —
13+ * surfaces sandbox-level defects (isolate crash, module load failure,
14+ * worker loader error) as a typed error so callers can handle them
15+ * structurally instead of untyped `unknown`. Runtimes that want a
16+ * narrower error shape can define their own `Data.TaggedError` subclass
17+ * and parameterize `CodeExecutor<MyError>`.
18+ */
19+export class CodeExecutionError extends Data.TaggedError("CodeExecutionError")<{
20+ readonly runtime: string;
21+ readonly message: string;
22+ readonly cause?: unknown;
23+}> {}
packages/kernel/core/src/types.ts+14−3
@@ -1,6 +1,9 @@
11 import type { StandardSchemaV1 } from "@standard-schema/spec";
2+import type * as Cause from "effect/Cause";
23 import type * as Effect from "effect/Effect";
34
5+import type { CodeExecutionError } from "./effect-errors";
6+
47 /** Branded tool path */
58 export type ToolPath = string & { readonly __toolPath: unique symbol };
69
@@ -30,9 +33,17 @@ export type ExecuteResult = { · 23 unmodified lines
3033 logs?: string[];
3134 };
3235
33−/** Executes code in a sandboxed runtime with tool access */
34−export interface CodeExecutor {
35− execute(code: string, toolInvoker: SandboxToolInvoker): Effect.Effect<ExecuteResult, unknown>;
36+/**
37+ * Executes code in a sandboxed runtime with tool access.
38+ *
39+ * Error channel is constrained to Effect's `YieldableError` (the base
40+ * shape `Data.TaggedError(...)` produces) so callers always get a
41+ * structurally tagged error, never untyped `unknown`. Defaults to
42+ * `CodeExecutionError`; runtimes can parameterize with their own
43+ * `Data.TaggedError` subclass — e.g. `CodeExecutor<WorkerLoaderError>`.
44+ */
45+export interface CodeExecutor<E extends Cause.YieldableError = CodeExecutionError> {
46+ execute(code: string, toolInvoker: SandboxToolInvoker): Effect.Effect<ExecuteResult, E>;
3647 }
3748
3849 /** Accept-anything schema for tools with no input validation */
packages/kernel/runtime-deno-subprocess/src/index.ts+1−1
@@ -330,7 +330,7 @@ export const isDenoAvailable = (executable: string = defaultDenoExecutable()): b · 329 unmodified lines
330330
331331 export const makeDenoSubprocessExecutor = (
332332 options: DenoSubprocessExecutorOptions = {},
333−): CodeExecutor => ({
333+): CodeExecutor<never> => ({
334334 execute: (code: string, toolInvoker: SandboxToolInvoker) =>
335335 executeInDeno(code, toolInvoker, options),
336336 });
packages/kernel/runtime-dynamic-worker/src/executor.ts+3−1
@@ -323,7 +323,9 @@ const runInDynamicWorker = ( · 322 unmodified lines
323323 // Public API
324324 // ---------------------------------------------------------------------------
325325
326−export const makeDynamicWorkerExecutor = (options: DynamicWorkerExecutorOptions): CodeExecutor => ({
326+export const makeDynamicWorkerExecutor = (
327+ options: DynamicWorkerExecutorOptions,
328+): CodeExecutor<DynamicWorkerExecutionError> => ({
327329 execute: (code: string, toolInvoker: SandboxToolInvoker) =>
328330 runInDynamicWorker(options, code, toolInvoker),
329331 });
packages/kernel/runtime-quickjs/src/index.ts+3−1
@@ -396,7 +396,9 @@ const runInQuickJs = ( · 395 unmodified lines
396396 }),
397397 );
398398
399−export const makeQuickJsExecutor = (options: QuickJsExecutorOptions = {}): CodeExecutor => ({
399+export const makeQuickJsExecutor = (
400+ options: QuickJsExecutorOptions = {},
401+): CodeExecutor<QuickJsExecutionError> => ({
400402 execute: (code: string, toolInvoker: SandboxToolInvoker) =>
401403 runInQuickJs(options, code, toolInvoker),
402404 });
packages/kernel/runtime-secure-exec/src/index.ts+3−1
@@ -441,7 +441,9 @@ const evaluateInSecureExec = ( · 440 unmodified lines
441441 );
442442 };
443443
444−export const makeSecureExecExecutor = (options: SecureExecExecutorOptions = {}): CodeExecutor => ({
444+export const makeSecureExecExecutor = (
445+ options: SecureExecExecutorOptions = {},
446+): CodeExecutor<never> => ({
445447 execute: (code: string, toolInvoker: SandboxToolInvoker) =>
446448 evaluateInSecureExec(options, code, toolInvoker),
447449 });