diff --git a/apps/cloud/src/api/protected.test.ts b/apps/cloud/src/api/protected.test.ts index f5a81f0ae5..fe0e523698 100644 --- a/apps/cloud/src/api/protected.test.ts +++ b/apps/cloud/src/api/protected.test.ts @@ -6,6 +6,7 @@ import { withExecutionUsageTracking } from "../engine/execution-usage"; const makeBaseEngine = (): ExecutionEngine => ({ execute: () => Effect.succeed({ result: "ok", logs: [] }), + call: () => Effect.succeed({ result: "ok", logs: [] }), executeWithPause: () => Effect.succeed({ status: "completed", diff --git a/apps/cloud/src/engine/execution-gate.ts b/apps/cloud/src/engine/execution-gate.ts index 21d6cfd1a3..7a51b2fd39 100644 --- a/apps/cloud/src/engine/execution-gate.ts +++ b/apps/cloud/src/engine/execution-gate.ts @@ -5,7 +5,8 @@ // Usage is tracked to Autumn after every execution (execution-usage.ts), but // nothing ever CHECKED the balance before running, so a free-tier org could run // unbounded executions past its quota. This gate consults -// `AutumnService.checkExecutionBalance` before `execute` / `executeWithPause`. +// `AutumnService.checkExecutionBalance` before `execute` / `call` / +// `executeWithPause`. // `resume` is never gated: a paused execution already consumed its quota slot // when it started, and blocking resume would strand approved work forever. // @@ -68,7 +69,7 @@ export type GateDecision = | { readonly blocked: true; readonly error: { readonly message: string } }; /** - * Wrap an engine so `decide` runs before `execute` / `executeWithPause`. A + * Wrap an engine so `decide` runs before `execute` / `call` / `executeWithPause`. A * blocked decision short-circuits to a descriptive `ExecuteResult.error` * (which `formatExecuteResult` renders as a clean `isError` MCP tool result) * WITHOUT invoking the inner engine — so a blocked execution is neither run @@ -87,6 +88,14 @@ export const withPreExecutionGate = ( ? Effect.succeed({ result: null, error: decision.error.message }) : engine.execute(code, options), ), + call: (address, args, options) => + Effect.flatMap( + decide, + (decision): Effect.Effect => + decision.blocked + ? Effect.succeed({ result: null, error: decision.error.message }) + : engine.call(address, args, options), + ), executeWithPause: (code, options) => Effect.flatMap( decide, diff --git a/apps/cloud/src/engine/execution-rate-limit.node.test.ts b/apps/cloud/src/engine/execution-rate-limit.node.test.ts index 4cefbdbf85..bef62f6021 100644 --- a/apps/cloud/src/engine/execution-rate-limit.node.test.ts +++ b/apps/cloud/src/engine/execution-rate-limit.node.test.ts @@ -25,6 +25,7 @@ class UpstreamDownError extends Data.TaggedError("UpstreamDownError")<{ // the two came back. A blocked decision never reaches the engine at all. const engineStub: ExecutionEngine = { execute: () => Effect.succeed({ result: "ran" }), + call: () => Effect.succeed({ result: "ran" }), executeWithPause: () => Effect.succeed({ status: "completed", result: { result: "ran" } }), resume: () => Effect.succeed(null), getPausedExecution: () => Effect.succeed(null), @@ -49,6 +50,27 @@ const runExecute = (limiter: ReturnType) => }), ); +describe("execution rate limiter: direct tool calls", () => { + it("counts and blocks a direct call like an execute", async () => { + const limiter = makeExecutionRateLimiter(countingIncrement([1, 11]), { + limit: 10, + isExempt: () => Effect.succeed(false), + }); + const engine = limiter.decorate(ORG, engineStub); + const call = () => + Effect.runPromise( + engine.call( + "tools.github.org.main.getRepo", + {}, + { onElicitation: () => Effect.die("elicitation is not exercised here") }, + ), + ); + + expect(await call()).toEqual({ result: "ran" }); + expect(await call()).toEqual({ result: null, error: RATE_LIMIT_BLOCKED_MESSAGE }); + }); +}); + describe("execution rate limiter — paid exemption", () => { it("allows executions under the cap without consulting the exemption", async () => { let exemptionCalls = 0; diff --git a/apps/cloud/src/engine/execution-usage.ts b/apps/cloud/src/engine/execution-usage.ts index d5b2bc8e99..e8f6e8a0a7 100644 --- a/apps/cloud/src/engine/execution-usage.ts +++ b/apps/cloud/src/engine/execution-usage.ts @@ -12,6 +12,10 @@ export const withExecutionUsageTracking = ( engine .execute(code, options) .pipe(Effect.tap(() => Effect.sync(() => trackUsage(organizationId)))), + call: (address, args, options) => + engine + .call(address, args, options) + .pipe(Effect.tap(() => Effect.sync(() => trackUsage(organizationId)))), // `options` MUST be forwarded: it carries `autoApprove`, and dropping it // silently turned every auto-approved execution back into a paused one. executeWithPause: (code, options) => diff --git a/apps/host-cloudflare/src/execution.ts b/apps/host-cloudflare/src/execution.ts index a63afea50b..f364d0f4f3 100644 --- a/apps/host-cloudflare/src/execution.ts +++ b/apps/host-cloudflare/src/execution.ts @@ -32,14 +32,20 @@ import { makeCloudflarePlugins } from "./plugins"; export { makeExecutionStack } from "@executor-js/api/server"; export { EngineDecoratorNoop }; +const workerLoader = (): WorkerLoader | undefined => (env as { LOADER?: WorkerLoader }).LOADER; + export const CloudflareCodeExecutorProvider: Layer.Layer = Layer.sync( CodeExecutorProvider, () => { - const loader = (env as { LOADER?: WorkerLoader }).LOADER; + const loader = workerLoader(); return loader ? makeDynamicWorkerExecutor({ loader }) : makeQuickJsExecutor(); }, ); +/** The code runtime `CloudflareCodeExecutorProvider` picks, for the MCP log line. */ +export const cloudflareCodeRuntime = (): "dynamic-worker" | "quickjs" => + workerLoader() ? "dynamic-worker" : "quickjs"; + export const makeCloudflarePluginsProvider = ( config: CloudflareConfig, ): Layer.Layer => diff --git a/apps/host-cloudflare/src/mcp/session-durable-object.ts b/apps/host-cloudflare/src/mcp/session-durable-object.ts index dd112494d0..a87f44a0ab 100644 --- a/apps/host-cloudflare/src/mcp/session-durable-object.ts +++ b/apps/host-cloudflare/src/mcp/session-durable-object.ts @@ -29,7 +29,11 @@ import type { ResumeResponse } from "@executor-js/execution"; import { loadConfig, type CloudflareConfig, type CloudflareEnv } from "../config"; import { createD1ExecutorDb } from "../db/d1"; -import { makeCloudflareExecutionStackLayer, makeExecutionStack } from "../execution"; +import { + cloudflareCodeRuntime, + makeCloudflareExecutionStackLayer, + makeExecutionStack, +} from "../execution"; // --------------------------------------------------------------------------- // Cloudflare (self-host) MCP Session Durable Object — the host-cloudflare @@ -162,6 +166,7 @@ export class McpSessionDO extends McpAgentSessionDOBase = { execute: () => Effect.succeed({ result: "unused" }), + call: () => Effect.succeed({ result: "unused" }), executeWithPause: () => Effect.succeed({ status: "completed", result: { result: "unused" } }), resume: () => Effect.succeed(null), getPausedExecution: () => Effect.succeed(null), diff --git a/packages/core/analytics/src/engine.test.ts b/packages/core/analytics/src/engine.test.ts index d3104fd7ba..8bdc4cc326 100644 --- a/packages/core/analytics/src/engine.test.ts +++ b/packages/core/analytics/src/engine.test.ts @@ -25,6 +25,8 @@ const makeRecorder = () => { }; }; +const declineAll = () => Effect.succeed({ action: "decline" as const }); + const completedResult = (error?: string): ExecutionResult => ({ status: "completed", result: { result: error ? null : "ok", ...(error ? { error } : {}) }, @@ -49,11 +51,13 @@ const pausedResult: ExecutionResult = { const makeFakeEngine = ( outcomes: Partial<{ execute: Effect.Effect<{ result: unknown; error?: string }, FakeExecutionError>; + call: Effect.Effect<{ result: unknown; error?: string }, FakeExecutionError>; executeWithPause: Effect.Effect; resume: Effect.Effect; }>, ): ExecutionEngine => ({ execute: () => outcomes.execute ?? Effect.succeed({ result: "ok" }), + call: () => outcomes.call ?? Effect.succeed({ result: "ok" }), executeWithPause: () => outcomes.executeWithPause ?? Effect.succeed(completedResult()), resume: () => outcomes.resume ?? Effect.succeed(null), getPausedExecution: () => Effect.succeed(null), @@ -145,6 +149,33 @@ describe("withExecutionAnalytics", () => { }), ); + it.effect("records each direct tool call as one execution", () => + Effect.gen(function* () { + const recorder = makeRecorder(); + const ok = withExecutionAnalytics(makeFakeEngine({}), recorder.analytics, { + plane: "mcp", + toolkit: false, + }); + yield* ok.call("tools.github.org.main.getRepo", {}, { onElicitation: declineAll }); + const failed = withExecutionAnalytics( + makeFakeEngine({ call: Effect.succeed({ result: null, error: "declined" }) }), + recorder.analytics, + { plane: "mcp", toolkit: false }, + ); + yield* failed.call("tools.github.org.main.getRepo", {}, { onElicitation: declineAll }); + expect(recorder.events).toEqual([ + { + name: "execution_completed", + properties: { ok: true, plane: "mcp", toolkit: false }, + }, + { + name: "execution_completed", + properties: { ok: false, plane: "mcp", toolkit: false }, + }, + ]); + }), + ); + it.effect("resume of an unknown execution records nothing", () => Effect.gen(function* () { const recorder = makeRecorder(); diff --git a/packages/core/analytics/src/engine.ts b/packages/core/analytics/src/engine.ts index 70796dfd07..f169bfcad0 100644 --- a/packages/core/analytics/src/engine.ts +++ b/packages/core/analytics/src/engine.ts @@ -49,6 +49,11 @@ export const withExecutionAnalytics = ( Effect.tap((result) => completed(!result.error)), Effect.tapError(() => completed(false)), ), + call: (address, args, options) => + engine.call(address, args, options).pipe( + Effect.tap((result) => completed(!result.error)), + Effect.tapError(() => completed(false)), + ), executeWithPause: (code, options) => engine.executeWithPause(code, options).pipe( Effect.tap(recordOutcome), diff --git a/packages/core/api/src/server/execution-stack-middleware.ts b/packages/core/api/src/server/execution-stack-middleware.ts index afe0b45958..25f626021d 100644 --- a/packages/core/api/src/server/execution-stack-middleware.ts +++ b/packages/core/api/src/server/execution-stack-middleware.ts @@ -339,13 +339,15 @@ class PlatformEngineUnavailable extends Data.TaggedError("PlatformEngineUnavaila * actually reach — `getPausedExecution`, via `GET /executions/:id` — answers * null (a 404), honestly: the platform view owns no executions. The paused * counters answer the same empty story for any future reader. Everything else - * (execute, resume, the MCP tool description) exists to satisfy the service + * (execute, call, resume, the MCP tool description) exists to satisfy the service * shape, sits behind the middleware's safe-request gate, and dies if reached. */ const readOnlyExecutionEngine: ExecutionEngine = { // oxlint-disable-next-line executor/no-effect-escape-hatch -- boundary: unreachable behind the middleware's safe-request gate; reaching it is a wiring bug, not a typed product outcome execute: () => Effect.die(new PlatformEngineUnavailable({ member: "execute" })), // oxlint-disable-next-line executor/no-effect-escape-hatch -- boundary: unreachable behind the middleware's safe-request gate; reaching it is a wiring bug, not a typed product outcome + call: () => Effect.die(new PlatformEngineUnavailable({ member: "call" })), + // oxlint-disable-next-line executor/no-effect-escape-hatch -- boundary: unreachable behind the middleware's safe-request gate; reaching it is a wiring bug, not a typed product outcome executeWithPause: () => Effect.die(new PlatformEngineUnavailable({ member: "executeWithPause" })), // oxlint-disable-next-line executor/no-effect-escape-hatch -- boundary: unreachable behind the middleware's safe-request gate; reaching it is a wiring bug, not a typed product outcome resume: () => Effect.die(new PlatformEngineUnavailable({ member: "resume" })), diff --git a/packages/core/execution/src/engine.ts b/packages/core/execution/src/engine.ts index dad73e2025..7c41fc9671 100644 --- a/packages/core/execution/src/engine.ts +++ b/packages/core/execution/src/engine.ts @@ -506,6 +506,22 @@ export type ExecutionEngine options: { readonly onElicitation: ElicitationHandler }, ) => Effect.Effect; + /** + * Call one tool by address in the host, with no sandbox. The result has the + * shape `execute` returns for `return await tools.(args)`: the tool's + * `ToolResult` on success, or a `tool_error` for a declined approval or a + * plugin defect. + * + * Required, not optional, for the same reason as `shutdown`: a decorator + * that dropped it would silently remove billing, gating, or analytics from + * direct calls. + */ + readonly call: ( + address: string, + args: unknown, + options: { readonly onElicitation: ElicitationHandler }, + ) => Effect.Effect; + /** * Execute code, intercepting the first elicitation as a pause point. * Use this when the host doesn't support inline elicitation. @@ -874,6 +890,39 @@ export const createExecutionEngine = toolPaths.push(path), + }); + const path = address.startsWith("tools.") ? address.slice("tools.".length) : address; + const result = yield* invoker.invoke({ path, args }).pipe( + Effect.map( + (value): ExecuteResult => ({ + result: value, + logs: [], + ...(toolPaths.length === 0 ? {} : { toolPaths }), + }), + ), + Effect.catchTag("ExecutionToolError", (error) => + Effect.succeed({ + result: null, + error: error.message, + errorKind: "tool_error", + logs: [], + }), + ), + ); + yield* annotateExecuteOutcome(result); + return result; + }); + /** * End this engine's sandbox fibers, and WAIT for them to finish unwinding. * @@ -898,6 +947,7 @@ export const createExecutionEngine = Promise; + readonly call: ( + address: string, + args: unknown, + options: { readonly onElicitation: ElicitationHandler }, + ) => Promise; readonly executeWithPause: ( code: string, options?: { readonly autoApprove?: boolean }, @@ -154,6 +159,14 @@ export const toPromiseExecutionEngine = ( Effect.tryPromise(() => options.onElicitation(ctx)).pipe(Effect.orDie), }), ), + call: (address, args, options) => + Effect.runPromise( + engine.call(address, args, { + onElicitation: (ctx) => + // oxlint-disable-next-line executor/no-effect-escape-hatch -- boundary: host-provided Promise elicitation callback is outside the Effect error model + Effect.tryPromise(() => options.onElicitation(ctx)).pipe(Effect.orDie), + }), + ), executeWithPause: (code, options) => Effect.runPromise(engine.executeWithPause(code, options)), resume: (executionId, response) => Effect.runPromise(engine.resume(executionId, response)), getPausedExecution: (executionId) => Effect.runPromise(engine.getPausedExecution(executionId)), diff --git a/packages/core/execution/src/tool-invoker.test.ts b/packages/core/execution/src/tool-invoker.test.ts index a2aa222d17..e5474318f0 100644 --- a/packages/core/execution/src/tool-invoker.test.ts +++ b/packages/core/execution/src/tool-invoker.test.ts @@ -2030,3 +2030,151 @@ describe("pause/resume with multiple elicitations", () => { }), ); }); + +// --------------------------------------------------------------------------- +// engine.call: one tool call in the host, with no sandbox +// --------------------------------------------------------------------------- + +const sandboxForbidden: CodeExecutor = { + execute: () => Effect.die("engine.call must not start a sandbox"), +}; + +const declineAll = () => Effect.succeed(ElicitationResponse.make({ action: "decline" })); + +const toolFile = { + _tag: "ToolFile", + name: "pixel.png", + mimeType: "image/png", + encoding: "base64", + data: "iVBORw0KGgo=", + byteLength: 8, +} as const; + +const binaryPlugin = makeTestPlugin({ + pluginId: "binary-test", + integration: "files", + tools: [ + { + name: "download", + description: "Download a file", + inputJsonSchema: EmptyInputJson, + validator: EmptyValidator, + handler: () => Effect.succeed(toolFile), + }, + { + name: "bytes", + description: "Read raw bytes", + inputJsonSchema: EmptyInputJson, + validator: EmptyValidator, + handler: () => Effect.succeed(new Uint8Array([1, 2, 3])), + }, + ], +}); + +describe("engine.call", () => { + it.effect("returns the tool's ToolResult and the connected tool path without a sandbox", () => + Effect.gen(function* () { + const executor = yield* makeSearchExecutor(); + const engine = createExecutionEngine({ executor, codeExecutor: sandboxForbidden }); + + const result = yield* engine.call( + "tools.github.org.main.getRepositoryDetails", + { owner: "octocat", repo: "hello" }, + { onElicitation: acceptAll }, + ); + + expect(result).toEqual({ + result: { ok: true, data: { defaultBranch: "main" } }, + logs: [], + toolPaths: ["github.org.main.getRepositoryDetails"], + }); + }), + ); + + it.effect("returns tool_blocked as a ToolResult when a block policy matches", () => + Effect.gen(function* () { + const executor = yield* makeSearchExecutor(); + yield* executor.policies.create({ owner: "org", pattern: "github.*", action: "block" }); + const engine = createExecutionEngine({ executor, codeExecutor: sandboxForbidden }); + + const result = yield* engine.call( + "tools.github.org.main.getRepositoryDetails", + { owner: "octocat", repo: "hello" }, + { onElicitation: acceptAll }, + ); + + expect(result.error).toBeUndefined(); + expect(result.toolPaths).toBeUndefined(); + expect(result.result).toMatchObject({ + ok: false, + error: { + code: "tool_blocked", + message: "Tool blocked by policy: github.org.main.getRepositoryDetails", + }, + }); + }), + ); + + it.effect("reports a declined approval as a tool_error with the sandbox's message", () => + Effect.gen(function* () { + const executor = yield* makeSearchExecutor(); + const engine = createExecutionEngine({ executor, codeExecutor: sandboxForbidden }); + + const result = yield* engine.call( + "tools.crm.org.main.createContact", + { email: "a@example.com" }, + { onElicitation: declineAll }, + ); + + expect(result).toEqual({ + result: null, + error: + 'Tool "crm.org.main.createContact" requires approval but the request was declined by the user.', + errorKind: "tool_error", + logs: [], + }); + }), + ); + + it.effect("keeps a plugin defect opaque behind a correlation id", () => + Effect.gen(function* () { + const executor = yield* makeExecutorWith([unmarkedErrorPlugin] as const); + yield* provision(executor as never, [ + { pluginId: "unmarked-error-test", integration: "unmarked" }, + ]); + const engine = createExecutionEngine({ executor, codeExecutor: sandboxForbidden }); + + const result = yield* engine.call( + "tools.unmarked.org.main.explode", + {}, + { onElicitation: acceptAll }, + ); + + expect(result.result).toBeNull(); + expect(result.errorKind).toBe("tool_error"); + expect(result.error).toMatch(/^Internal tool error \[[0-9a-f]{8}\]$/); + expect(result.error).not.toContain("internal detail"); + }), + ); + + it.effect("returns a ToolFile and raw bytes unchanged", () => + Effect.gen(function* () { + const executor = yield* makeExecutorWith([binaryPlugin] as const); + yield* provision(executor as never, [{ pluginId: "binary-test", integration: "files" }]); + const engine = createExecutionEngine({ executor, codeExecutor: sandboxForbidden }); + + const file = yield* engine.call( + "tools.files.org.main.download", + {}, + { + onElicitation: acceptAll, + }, + ); + expect(file.result).toEqual({ ok: true, data: toolFile }); + + const bytes = yield* engine.call("files.org.main.bytes", {}, { onElicitation: acceptAll }); + expect(bytes.result).toEqual({ ok: true, data: new Uint8Array([1, 2, 3]) }); + expect(bytes.toolPaths).toEqual(["files.org.main.bytes"]); + }), + ); +}); diff --git a/packages/core/execution/src/tool-invoker.ts b/packages/core/execution/src/tool-invoker.ts index 0881d5ac3f..47487d3873 100644 --- a/packages/core/execution/src/tool-invoker.ts +++ b/packages/core/execution/src/tool-invoker.ts @@ -311,13 +311,17 @@ const extractNamespace = (path: string): string => { * because it would require an `integrations.list()` lookup on every invocation. * Callers that already know the integration kind can annotate at their own span. */ +export interface ExecutorToolInvoker extends SandboxToolInvoker { + invoke(input: { path: string; args: unknown }): Effect.Effect; +} + export const makeExecutorToolInvoker = ( executor: Executor, options: { readonly invokeOptions: InvokeOptions; readonly onConnectedToolCall?: (path: string) => void; }, -): SandboxToolInvoker => ({ +): ExecutorToolInvoker => ({ invoke: Effect.fn("mcp.tool.dispatch")(function* ({ path, args }) { yield* Effect.annotateCurrentSpan({ "mcp.tool.name": path, diff --git a/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.test.ts b/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.test.ts index 813f4d5c40..f24fa09e46 100644 --- a/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.test.ts +++ b/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.test.ts @@ -262,6 +262,7 @@ const makeEngine = ( calls, engine: { execute: () => Effect.succeed({ result: "execute-result" }), + call: () => Effect.succeed({ result: "execute-result" }), executeWithPause: () => Effect.succeed(completed("execute-result")), resume: (executionId, response) => Effect.sync(() => { diff --git a/packages/hosts/cloudflare/src/mcp/agent-session-model-resume.test.ts b/packages/hosts/cloudflare/src/mcp/agent-session-model-resume.test.ts index 7c8128c8b0..137377a396 100644 --- a/packages/hosts/cloudflare/src/mcp/agent-session-model-resume.test.ts +++ b/packages/hosts/cloudflare/src/mcp/agent-session-model-resume.test.ts @@ -265,6 +265,7 @@ const makeEngine = ( ); const engine: ExecutionEngine = { execute: () => Effect.succeed({ result: "execute-result" }), + call: () => Effect.succeed({ result: "execute-result" }), executeWithPause: () => Effect.succeed(completed("execute-result")), resume, getPausedExecution: () => Effect.succeed(null), diff --git a/packages/hosts/mcp-apps-shell/src/shell-resource.smoke.test.ts b/packages/hosts/mcp-apps-shell/src/shell-resource.smoke.test.ts index 83cac84946..867b9a1ed6 100644 --- a/packages/hosts/mcp-apps-shell/src/shell-resource.smoke.test.ts +++ b/packages/hosts/mcp-apps-shell/src/shell-resource.smoke.test.ts @@ -21,6 +21,7 @@ import { MCP_APPS_SHELL_DOCUMENT_MARKER } from "./shell-asset-path"; const stubEngine: ExecutionEngine = { execute: () => Effect.succeed({ result: "ok" }), + call: () => Effect.succeed({ result: "ok" }), executeWithPause: () => Effect.succeed({ status: "completed", result: { result: "ok" } }), resume: () => Effect.succeed(null), isExecutionSettled: undefined, diff --git a/packages/hosts/mcp-apps-shell/src/shell/mcp-app.browser.test.ts b/packages/hosts/mcp-apps-shell/src/shell/mcp-app.browser.test.ts index a41cd05c68..26e04d3513 100644 --- a/packages/hosts/mcp-apps-shell/src/shell/mcp-app.browser.test.ts +++ b/packages/hosts/mcp-apps-shell/src/shell/mcp-app.browser.test.ts @@ -1425,6 +1425,7 @@ const startSchemaElicitationMcpHarness = (): Promise => // The fake forks nothing, so there is no sandbox fiber to end. shutdown: Effect.void, execute: () => Effect.succeed({ result: null }), + call: () => Effect.succeed({ result: null }), executeWithPause: () => Effect.succeed( makePausedResult( diff --git a/packages/hosts/mcp/src/artifacts-tools.test.ts b/packages/hosts/mcp/src/artifacts-tools.test.ts index a761f12bb8..fbf9139b54 100644 --- a/packages/hosts/mcp/src/artifacts-tools.test.ts +++ b/packages/hosts/mcp/src/artifacts-tools.test.ts @@ -45,6 +45,7 @@ const makeStubEngine = (overrides: { resume?: ExecutionEngine["resume"]; }): ExecutionEngine => ({ execute: () => Effect.succeed({ result: "default" }), + call: () => Effect.succeed({ result: "default" }), executeWithPause: overrides.executeWithPause ?? (() => Effect.succeed({ status: "completed", result: { result: "default" } })), diff --git a/packages/hosts/mcp/src/in-memory-session-store.test.ts b/packages/hosts/mcp/src/in-memory-session-store.test.ts index bdc51db85d..81ab7443a7 100644 --- a/packages/hosts/mcp/src/in-memory-session-store.test.ts +++ b/packages/hosts/mcp/src/in-memory-session-store.test.ts @@ -63,6 +63,7 @@ it("preserves native elicitation mode when creating an in-memory MCP session", a /** A do-nothing engine: the eviction test drives session lifetime, not tools. */ const makeIdleTestEngine = (): ExecutionEngine => ({ execute: () => Effect.succeed({ result: "unused" }), + call: () => Effect.succeed({ result: "unused" }), executeWithPause: () => Effect.succeed({ status: "completed", result: { result: "unused" } }), resume: () => Effect.succeed(null), getPausedExecution: () => Effect.succeed(null), @@ -558,6 +559,7 @@ const MCP_POST_HEADERS = { /** No code ever runs here: these requests are answered before any tool call. */ const stubEngine: ExecutionEngine = { execute: () => Effect.succeed({ result: "unused" }), + call: () => Effect.succeed({ result: "unused" }), executeWithPause: () => Effect.succeed({ status: "completed", result: { result: "unused" } }), resume: () => Effect.succeed(null), getPausedExecution: () => Effect.succeed(null), diff --git a/packages/hosts/mcp/src/namespace-search-tools.test.ts b/packages/hosts/mcp/src/namespace-search-tools.test.ts index 428fb091f9..2ab915cddf 100644 --- a/packages/hosts/mcp/src/namespace-search-tools.test.ts +++ b/packages/hosts/mcp/src/namespace-search-tools.test.ts @@ -24,6 +24,7 @@ const makeRecordingEngine = (): { executed, engine: { execute: () => Effect.succeed({ result: "default" }), + call: () => Effect.succeed({ result: "default" }), executeWithPause: (code) => Effect.sync(() => { executed.push(code); diff --git a/packages/hosts/mcp/src/passthrough-tools.test.ts b/packages/hosts/mcp/src/passthrough-tools.test.ts index 9bea56b141..2e2bca816b 100644 --- a/packages/hosts/mcp/src/passthrough-tools.test.ts +++ b/packages/hosts/mcp/src/passthrough-tools.test.ts @@ -4,8 +4,11 @@ import { Client } from "@modelcontextprotocol/sdk/client/index.js"; import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js"; import type * as Cause from "effect/Cause"; -import type { ExecutionEngine } from "@executor-js/execution"; +import { createExecutionEngine, type ExecutionEngine } from "@executor-js/execution"; import { + createExecutor, + definePlugin, + tool, ToolAddress, IntegrationSlug, ConnectionName, @@ -13,9 +16,9 @@ import { type Tool, type ToolSchemaView, } from "@executor-js/sdk"; +import { makeTestConfig } from "@executor-js/sdk/testing"; import { readToolMode } from "./browser-approval"; -import { passthroughCallCode } from "./passthrough-tools"; import { createExecutorMcpServer, McpPassthroughUnavailableError, @@ -86,11 +89,11 @@ const toolPort = ( }), }); -/** A stub engine that records every executed code string and answers with a - * fixed value, so a test can prove a passthrough call became the expected - * single-call code and took `execute` (never `executeWithPause`). */ +/** A stub engine that records every direct call and executed code string and + * answers with a fixed value, so a test can prove a passthrough call became + * one direct `call` (never `execute` or `executeWithPause`). */ const makeRecordingEngine = (result: unknown = { ok: true, data: { hello: "world" } }) => { - const executed: string[] = []; + const executed: unknown[] = []; let pausedCalls = 0; const engine: ExecutionEngine = { execute: (code) => @@ -98,6 +101,11 @@ const makeRecordingEngine = (result: unknown = { ok: true, data: { hello: "world executed.push(code); return { result }; }), + call: (address, args) => + Effect.sync(() => { + executed.push({ address, args }); + return { result }; + }), executeWithPause: () => Effect.sync(() => { pausedCalls += 1; @@ -138,11 +146,6 @@ const withClient = async ( } }; -const decodeJsonString = Schema.decodeUnknownSync(Schema.fromJsonString(Schema.String)); -const decodeJsonRecord = Schema.decodeUnknownSync( - Schema.fromJsonString(Schema.Record(Schema.String, Schema.Unknown)), -); - const decodeSearchItems = Schema.decodeUnknownSync( Schema.Struct({ items: Schema.Array(Schema.Struct({ id: Schema.String })) }), ); @@ -163,37 +166,6 @@ const CATALOG: readonly Tool[] = [ projection({ integration: "linear", name: "issueCreate", requiresApproval: true }), ]; -describe("passthrough catalog", () => { - it("emits exactly one awaited tool call with the whole address as one string literal", () => { - expect(passthroughCallCode("tools.github.org.main.issues.create", { title: "hi" })).toBe( - 'return await tools["github.org.main.issues.create"]({"title":"hi"});', - ); - expect(passthroughCallCode("linear.org.main.issueCreate", undefined)).toBe( - 'return await tools["linear.org.main.issueCreate"]({});', - ); - // `then` is reserved by every sandbox proxy; as part of one key it is - // just text, so such a tool stays callable. - expect(passthroughCallCode("tools.svc.org.main.items.then", {})).toBe( - 'return await tools["svc.org.main.items.then"]({});', - ); - }); - - it("keeps a hostile tool segment as data, never as code", () => { - // An OpenAPI spec controls its tool paths (`x-executor-toolPath`), so a - // segment can contain anything. It must land inside a JSON string. - const hostile = 'x"](await tools.victim.org.main.destroy({}))["'; - const code = passthroughCallCode(`tools.evil.org.main.${hostile}`, {}); - // Structural proof the payload never escapes the string literal: the - // source is exactly `return await tools[]();`, and - // that one string decodes back to the raw address. - const shape = /^return await tools\[("(?:[^"\\]|\\.)*")\]\((\{.*\})\);$/s.exec(code); - expect(shape).not.toBeNull(); - expect(decodeJsonString(shape![1]!)).toBe(`evil.org.main.${hostile}`); - // And the call's argument is the JSON we passed, untouched by the address. - expect(decodeJsonRecord(shape![2]!)).toEqual({}); - }); -}); - // --------------------------------------------------------------------------- // Wire flags // --------------------------------------------------------------------------- @@ -468,7 +440,7 @@ describe("passthrough mode server", () => { arguments: { tool: found?.id, arguments: {} }, }); expect(invoked.isError ?? false).toBe(false); - expect(executed).toEqual(['return await tools["bench.org.main.record9999"]({});']); + expect(executed).toEqual([{ address: "tools.bench.org.main.record9999", args: {} }]); const page = await client.callTool({ name: "search", arguments: { query: "bench", limit: 2 }, @@ -549,7 +521,10 @@ describe("passthrough mode server", () => { }); expect(unknown.isError).toBe(true); expect(recording.executed).toEqual([ - 'return await tools["executor.mcp.addServer"]({"endpoint":"https://blog.mcp.cloudflare.com/mcp"});', + { + address: "executor.mcp.addServer", + args: { endpoint: "https://blog.mcp.cloudflare.com/mcp" }, + }, ]); }, ); @@ -622,7 +597,7 @@ describe("passthrough mode server", () => { }); }); - it("runs a call as one execute of single-call code, never a pause", async () => { + it("runs a call as one direct tool call, never code or a pause", async () => { const recording = makeRecordingEngine({ ok: true, data: { number: 7 } }); await withClient( { @@ -636,7 +611,7 @@ describe("passthrough mode server", () => { arguments: { tool: "tools.github.org.main.issues.create", arguments: { title: "hello" } }, }); expect(recording.executed).toEqual([ - 'return await tools["github.org.main.issues.create"]({"title":"hello"});', + { address: "tools.github.org.main.issues.create", args: { title: "hello" } }, ]); expect(recording.pausedCalls()).toBe(0); // The tool's `data` is the result: nothing sits between the tool and @@ -760,7 +735,7 @@ describe("passthrough mode server", () => { seen: string[], ): ExecutionEngine => ({ ...makeRecordingEngine().engine, - execute: (_code, options) => + call: (_address, _args, options) => Effect.gen(function* () { for (const { source, request } of requests) { const answer = yield* options.onElicitation({ @@ -1031,3 +1006,117 @@ describe("passthrough mode server", () => { expect(outcome).toBeInstanceOf(McpPassthroughUnavailableError); }); }); + +describe("passthrough invoke on a real engine", () => { + const pixel = { + _tag: "ToolFile", + name: "pixel.png", + mimeType: "image/png", + encoding: "base64", + data: "iVBORw0KGgo=", + byteLength: 8, + } as const; + + const emptyInput = Schema.toStandardSchemaV1(Schema.toStandardJSONSchemaV1(Schema.Struct({}))); + + const pixelsPlugin = definePlugin(() => ({ + id: "pixels-test" as const, + storage: () => ({}), + staticIntegrations: () => [ + { + id: "pixels", + kind: "in-memory" as const, + name: "Pixels", + tools: [ + tool({ + name: "download", + description: "Download a pixel", + inputSchema: emptyInput, + execute: () => Effect.succeed(pixel), + }), + tool({ + name: "explode", + description: "Fail with an internal detail", + inputSchema: emptyInput, + execute: () => Effect.die("postgres://secret@internal"), + }), + ], + }, + ], + })); + + const withRealEngine = async (fn: (client: Client) => Promise) => { + const executor = await Effect.runPromise( + createExecutor(makeTestConfig({ plugins: [pixelsPlugin()] as const })), + ); + const engine = createExecutionEngine({ + executor, + codeExecutor: { execute: () => Effect.die("invoke must not start a sandbox") }, + }); + // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: test helper must close the executor after async client assertions + try { + await withClient( + { + engine, + mode: "passthrough", + tools: executor.tools, + connections: executor.connections, + integrations: executor.integrations, + }, + fn, + ); + } finally { + await Effect.runPromise(executor.close()); + } + }; + + it("calls the tool in the host and renders a file result as before", async () => { + await withRealEngine(async (client) => { + const result = await client.callTool({ + name: "invoke", + arguments: { tool: "pixels.download", arguments: {} }, + }); + expect(result.isError ?? false).toBe(false); + expect(result.content).toEqual([ + { + type: "text", + text: [ + "{", + ' "_tag": "ToolFile",', + ' "name": "pixel.png",', + ' "mimeType": "image/png",', + ' "encoding": "base64",', + ' "data": "iVBORw0KGgo=",', + ' "byteLength": 8', + "}", + ].join("\n"), + }, + ]); + expect(result.structuredContent).toEqual({ + status: "completed", + result: pixel, + logs: [], + }); + }); + }); + + it("renders a plugin defect as an opaque tool error", async () => { + await withRealEngine(async (client) => { + const result = await client.callTool({ + name: "invoke", + arguments: { tool: "pixels.explode", arguments: {} }, + }); + expect(result.isError).toBe(true); + const text = (result.content as Array<{ text?: string }>)[0]?.text ?? ""; + expect(text).toMatch(/^Error: Internal tool error \[[0-9a-f]{8}\]$/); + // The engine's typed tool_error (with the sandbox's empty logs), not + // the host's catch-all failure, which carries no logs. + expect(result.structuredContent).toEqual({ + status: "error", + error: text.slice("Error: ".length), + logs: [], + }); + expect(JSON.stringify(result)).not.toContain("postgres://"); + }); + }); +}); diff --git a/packages/hosts/mcp/src/passthrough-tools.ts b/packages/hosts/mcp/src/passthrough-tools.ts index 7fd79a0753..f847ced183 100644 --- a/packages/hosts/mcp/src/passthrough-tools.ts +++ b/packages/hosts/mcp/src/passthrough-tools.ts @@ -1,28 +1,5 @@ import type { Skill } from "@executor-js/execution"; -/** - * The sandbox code a passthrough call runs. Built HERE from the session's - * resolved address and a JSON-encoded argument — never concatenated from raw - * model input — and shaped exactly like the artifact `execute-action` grammar - * (`return await tools.()`), so it takes the same engine path as - * every other execution: billing, rate limits, shape memory and analytics all - * see it as one execution. - */ -export const passthroughCallCode = (address: string, args: unknown): string => { - // The whole dotted address is ONE JSON string literal in bracket notation: - // `tools["github.org.main.items.then"](...)`. Two reasons it is not a chain - // of property accesses. The tool segment is customer-controlled (an OpenAPI - // spec may set `x-executor-toolPath`), so it must be data in the generated - // source, never syntax. And every sandbox proxy reserves the property name - // `then` (a thenable check would otherwise await the proxy itself), so a - // per-segment chain could never reach a tool whose path contains `then`. - // Each proxy joins the accessed keys with `.` to form the dispatch path, so - // a single key holding the dotted address reassembles to exactly the same - // path the chain would have. - const bare = address.startsWith("tools.") ? address.slice("tools.".length) : address; - return `return await tools[${JSON.stringify(bare)}](${JSON.stringify(args ?? {})});`; -}; - /** Describe the fixed search/invoke surface without listing the underlying catalog. */ export const passthroughInstructions = (): string => 'Use integrations to see connected accounts and skills({ name: "search-invoke" }) for the workflow. ' + diff --git a/packages/hosts/mcp/src/tool-call-log.test.ts b/packages/hosts/mcp/src/tool-call-log.test.ts new file mode 100644 index 0000000000..105258a10d --- /dev/null +++ b/packages/hosts/mcp/src/tool-call-log.test.ts @@ -0,0 +1,263 @@ +import { describe, expect, it } from "@effect/vitest"; +import { Effect, Logger } from "effect"; +import { Client } from "@modelcontextprotocol/sdk/client/index.js"; +import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js"; + +import type { ExecutionEngine, ExecutionResult } from "@executor-js/execution"; +import { + ConnectionName, + IntegrationSlug, + ToolAddress, + ToolName, + type Tool, + type ToolSchemaView, +} from "@executor-js/sdk"; + +import { createExecutorMcpServer, type McpToolsPort } from "./tool-server"; + +interface LogLine { + readonly level: string; + readonly message: unknown; +} + +const capturingLogger = (lines: LogLine[]) => + Logger.make((options) => { + lines.push({ level: options.logLevel, message: options.message }); + }); + +/** The fields of every "executor mcp tool call" line, in order. */ +function toolCallFields(lines: readonly LogLine[]): Record[] { + return lines.flatMap((line) => { + const parts = Array.isArray(line.message) ? line.message : [line.message]; + if (parts[0] !== "executor mcp tool call") return []; + expect(line.level).toBe("Info"); + return [parts[1] as Record]; + }); +} + +const githubTool: Tool = { + address: ToolAddress.make("tools.github.org.main.issues.list"), + integration: IntegrationSlug.make("github"), + owner: "org", + connection: ConnectionName.make("main"), + name: ToolName.make("issues.list"), + pluginId: "test", + description: "List issues", + inputSchema: { type: "object" }, +}; + +const toolPort: McpToolsPort = { + list: () => Effect.succeed([githubTool]), + schema: (address) => + Effect.succeed( + address === githubTool.address + ? ({ + address, + name: githubTool.name, + inputSchema: { type: "object" }, + } satisfies ToolSchemaView) + : null, + ), +}; + +function stubEngine(overrides: Partial): ExecutionEngine { + return { + execute: () => Effect.succeed({ result: null }), + call: () => Effect.succeed({ result: null }), + executeWithPause: () => Effect.succeed({ status: "completed", result: { result: null } }), + resume: () => Effect.succeed(null), + getPausedExecution: () => Effect.succeed(null), + pausedExecutionCount: () => Effect.succeed(0), + hasPausedExecutions: () => Effect.succeed(false), + getDescription: Effect.succeed("test executor"), + shutdown: Effect.void, + ...overrides, + }; +} + +async function callTool( + config: { + readonly engine: ExecutionEngine; + readonly mode?: "passthrough"; + readonly elicitationMode?: { readonly mode: "model" }; + }, + calls: ReadonlyArray<{ readonly name: string; readonly arguments: Record }>, +): Promise[]> { + const lines: LogLine[] = []; + const mcpServer = await Effect.runPromise( + createExecutorMcpServer({ + ...config, + codeRuntime: "dynamic-worker", + tools: toolPort, + connections: { list: () => Effect.succeed([]) }, + integrations: { list: () => Effect.succeed([]) }, + }).pipe(Effect.provide(Logger.layer([capturingLogger(lines)]))), + ); + const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair(); + const client = new Client({ name: "probe-client", version: "2.3.4" }, { capabilities: {} }); + await mcpServer.connect(serverTransport); + await client.connect(clientTransport); + // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: test helper must close MCP transports after async client assertions + try { + for (const call of calls) await client.callTool(call); + } finally { + await clientTransport.close(); + await serverTransport.close(); + } + return toolCallFields(lines); +} + +const SECRET_CODE = 'return await tools.github.org.main.issues.list({ token: "sk-secret" })'; + +describe("one log line per MCP tool call", () => { + it("logs a passthrough invoke as a direct call, without arguments or results", async () => { + const engine = stubEngine({ + call: () => + Effect.succeed({ + result: { ok: true, data: { secretResult: "do-not-log" } }, + toolPaths: ["github.org.main.issues.list"], + }), + }); + const fields = await callTool({ engine, mode: "passthrough" }, [ + { + name: "invoke", + arguments: { tool: "tools.github.org.main.issues.list", arguments: { q: "sk-secret" } }, + }, + ]); + expect(fields).toEqual([ + { + tool: "invoke", + runtime: "direct", + mode: "passthrough", + client: "probe-client/2.3.4", + toolCalls: 1, + discoveryOnly: false, + durationMs: expect.any(Number), + outcome: "ok", + }, + ]); + expect(JSON.stringify(fields)).not.toContain("sk-secret"); + expect(JSON.stringify(fields)).not.toContain("do-not-log"); + }); + + it("logs a failed passthrough tool result as a tool_error", async () => { + const engine = stubEngine({ + call: () => + Effect.succeed({ result: { ok: false, error: { code: "tool_blocked", message: "x" } } }), + }); + const fields = await callTool({ engine, mode: "passthrough" }, [ + { name: "invoke", arguments: { tool: "tools.github.org.main.issues.list", arguments: {} } }, + ]); + expect(fields).toMatchObject([{ tool: "invoke", toolCalls: 0, outcome: "tool_error" }]); + }); + + it("logs an execute that called no connected tool as discovery only", async () => { + const code = 'return await tools.search({ query: "issues" })'; + const engine = stubEngine({ + executeWithPause: () => + Effect.succeed({ status: "completed", result: { result: { items: [] } } }), + }); + const fields = await callTool({ engine }, [{ name: "execute", arguments: { code } }]); + expect(fields).toEqual([ + { + tool: "execute", + runtime: "dynamic-worker", + mode: "code", + client: "probe-client/2.3.4", + toolCalls: 0, + discoveryOnly: true, + codeLength: code.length, + durationMs: expect.any(Number), + outcome: "ok", + }, + ]); + }); + + it("counts the connected tool calls of an execute and never logs its code", async () => { + const engine = stubEngine({ + executeWithPause: () => + Effect.succeed({ + status: "completed", + result: { + result: null, + toolPaths: ["github.org.main.issues.list", "github.org.main.issues.list"], + }, + }), + }); + const fields = await callTool({ engine }, [ + { name: "execute", arguments: { code: SECRET_CODE } }, + ]); + expect(fields).toMatchObject([ + { tool: "execute", toolCalls: 2, discoveryOnly: false, outcome: "ok" }, + ]); + expect(JSON.stringify(fields)).not.toContain("sk-secret"); + }); + + it("logs the outcome of a failed, a thrown, and a paused execute", async () => { + const paused: ExecutionResult = { + status: "paused", + execution: { + id: "exec_1", + elicitationContext: { + address: githubTool.address, + args: {}, + request: { _tag: "FormElicitation", message: "Approve?", requestedSchema: {} }, + }, + }, + }; + const outcomes: ExecutionResult[] = [ + { + status: "completed", + result: { result: null, error: "declined", errorKind: "tool_error" }, + }, + { status: "completed", result: { result: null, error: "boom", errorKind: "thrown" } }, + paused, + ]; + const engine = stubEngine({ + executeWithPause: () => Effect.sync(() => outcomes.shift()!), + }); + const fields = await callTool({ engine, elicitationMode: { mode: "model" } }, [ + { name: "execute", arguments: { code: "return 1" } }, + { name: "execute", arguments: { code: "return 2" } }, + { name: "execute", arguments: { code: "return 3" } }, + ]); + expect(fields.map((field) => [field.outcome, field.discoveryOnly])).toEqual([ + ["tool_error", false], + ["error", false], + ["paused", false], + ]); + }); + + it("logs a resume with the runtime that continued the execution", async () => { + const engine = stubEngine({ + resume: () => + Effect.succeed({ + status: "completed", + result: { result: null, toolPaths: ["github.org.main.issues.list"] }, + }), + }); + const fields = await callTool({ engine, elicitationMode: { mode: "model" } }, [ + { name: "resume", arguments: { executionId: "exec_1", action: "accept" } }, + ]); + expect(fields).toEqual([ + { + tool: "resume", + runtime: "dynamic-worker", + mode: "code", + client: "probe-client/2.3.4", + toolCalls: 1, + discoveryOnly: false, + durationMs: expect.any(Number), + outcome: "ok", + }, + ]); + }); + + it("logs an engine failure as an error", async () => { + const engine = stubEngine({ executeWithPause: () => Effect.die("sandbox down") }); + const fields = await callTool({ engine }, [{ name: "execute", arguments: { code: "x" } }]); + expect(fields).toMatchObject([ + { tool: "execute", runtime: "dynamic-worker", discoveryOnly: false, outcome: "error" }, + ]); + }); +}); diff --git a/packages/hosts/mcp/src/tool-call-log.ts b/packages/hosts/mcp/src/tool-call-log.ts new file mode 100644 index 0000000000..a2534f5d4f --- /dev/null +++ b/packages/hosts/mcp/src/tool-call-log.ts @@ -0,0 +1,95 @@ +import { Context, Effect, Exit } from "effect"; + +import type { ExecutionResult } from "@executor-js/execution"; +import { isToolResult } from "@executor-js/sdk"; + +type ExecuteResult = Extract["result"]; + +/** The code runtime the engine's code executor uses. `sandbox` when the host + * did not say which one. */ +export type McpCodeRuntime = "dynamic-worker" | "quickjs" | "sandbox"; + +export type McpToolCallRuntime = McpCodeRuntime | "direct" | "none"; + +export type McpLoggedTool = "execute" | "invoke" | "execute_action" | "search_namespace" | "resume"; + +type McpToolCallOutcome = "ok" | "tool_error" | "error" | "paused"; + +/** What the engine did for one MCP tool call. Written by `trackEngineRun`, + * read once when the call ends. */ +interface EngineRun { + runtime: McpToolCallRuntime; + result?: ExecutionResult; +} + +const CurrentEngineRun = Context.Reference("executor/mcp/CurrentEngineRun", { + defaultValue: () => null, +}); + +/** Record that `engine` ran on `runtime`, and what it returned. */ +export const trackEngineRun = ( + runtime: McpToolCallRuntime, + engine: Effect.Effect, +): Effect.Effect => + Effect.gen(function* () { + const run = yield* CurrentEngineRun; + if (run) run.runtime = runtime; + const result = yield* engine; + if (run && result) { + run.result = "status" in result ? result : { status: "completed", result }; + } else if (run) { + run.runtime = "none"; + } + return result; + }); + +const outcomeOf = (run: EngineRun, exit: Exit.Exit<{ readonly isError?: boolean }, unknown>) => { + if (Exit.isFailure(exit)) return "error" satisfies McpToolCallOutcome; + if (run.result?.status === "paused") return "paused" satisfies McpToolCallOutcome; + const result = run.result?.result; + if (!result) return exit.value.isError ? "error" : "ok"; + if (result.error) return result.errorKind === "tool_error" ? "tool_error" : "error"; + if (run.runtime === "direct" && isToolResult(result.result) && !result.result.ok) { + return "tool_error"; + } + return "ok"; +}; + +/** + * Run one MCP tool call and then emit exactly one `executor mcp tool call` + * info line: which tool, where it ran, how many connected tools it called, + * and how it ended. It never carries arguments, code, results, or secrets. + */ +export const logToolCall = ( + input: { + readonly tool: McpLoggedTool; + readonly client: () => string | undefined; + readonly codeLength?: number; + }, + effect: Effect.Effect, +): Effect.Effect => + Effect.suspend(() => { + const startedAt = Date.now(); + const run: EngineRun = { runtime: "none" }; + return effect.pipe( + Effect.provideService(CurrentEngineRun, run), + Effect.onExit((exit) => { + const outcome = outcomeOf(run, exit); + const toolCalls = + run.result?.status === "completed" ? (run.result.result.toolPaths?.length ?? 0) : 0; + const ranCode = run.runtime !== "direct" && run.runtime !== "none"; + const client = input.client(); + return Effect.logInfo("executor mcp tool call", { + tool: input.tool, + runtime: run.runtime, + mode: input.tool === "invoke" ? "passthrough" : "code", + ...(client === undefined ? {} : { client }), + toolCalls, + discoveryOnly: ranCode && outcome === "ok" && toolCalls === 0, + ...(input.codeLength === undefined ? {} : { codeLength: input.codeLength }), + durationMs: Date.now() - startedAt, + outcome, + }); + }), + ); + }); diff --git a/packages/hosts/mcp/src/tool-server.test.ts b/packages/hosts/mcp/src/tool-server.test.ts index 04fd4fca28..9efba2049b 100644 --- a/packages/hosts/mcp/src/tool-server.test.ts +++ b/packages/hosts/mcp/src/tool-server.test.ts @@ -41,12 +41,14 @@ const expectDefined: (value: T) => asserts value is NonNullable = (value) const makeStubEngine = (overrides: { execute?: ExecutionEngine["execute"]; + call?: ExecutionEngine["call"]; executeWithPause?: ExecutionEngine["executeWithPause"]; resume?: ExecutionEngine["resume"]; isExecutionSettled?: ExecutionEngine["isExecutionSettled"]; description?: string; }): ExecutionEngine => ({ execute: overrides.execute ?? (() => Effect.succeed({ result: "default" })), + call: overrides.call ?? (() => Effect.succeed({ result: "default" })), executeWithPause: overrides.executeWithPause ?? (() => Effect.succeed({ status: "completed", result: { result: "default" } })), diff --git a/packages/hosts/mcp/src/tool-server.ts b/packages/hosts/mcp/src/tool-server.ts index caffc91ddf..335559465b 100644 --- a/packages/hosts/mcp/src/tool-server.ts +++ b/packages/hosts/mcp/src/tool-server.ts @@ -88,12 +88,9 @@ import { type BindableConnection, } from "./artifact-bindings"; import { MCP_ORG_WRITE_ACCESS_HEADER } from "./seams"; -import { - passthroughCallCode, - passthroughInstructions, - SEARCH_INVOKE_SKILL, -} from "./passthrough-tools"; +import { passthroughInstructions, SEARCH_INVOKE_SKILL } from "./passthrough-tools"; import type { McpToolMode } from "./browser-approval"; +import { logToolCall, trackEngineRun, type McpCodeRuntime } from "./tool-call-log"; // --------------------------------------------------------------------------- // Workers-compatible JSON Schema validator (replaces Ajv which uses new Function()) @@ -213,6 +210,11 @@ type SharedMcpServerConfig = { * Invoke is marked destructive for client approval. Requires `tools`. */ readonly mode?: McpToolMode; + /** + * The code runtime behind the engine, reported on the per-call log line. + * Defaults to `sandbox` when the host does not say. + */ + readonly codeRuntime?: McpCodeRuntime; /** * Ranks `search`. Passed explicitly because a caller that supplies a prebuilt * `engine` gives this server no way to reach the engine's own provider, and @@ -1659,7 +1661,7 @@ export const createExecutorMcpServer = ( const resumeWithLifecycle = (executionId: string, response: ResumeResponse) => Effect.gen(function* () { yield* onResumeStarted(executionId); - return yield* engine.resume(executionId, response); + return yield* trackEngineRun(codeRuntime, engine.resume(executionId, response)); }).pipe(Effect.ensuring(onResumeSettled(executionId))); const localExecutionAlreadySettled = (executionId: string): Effect.Effect => @@ -1730,15 +1732,22 @@ export const createExecutorMcpServer = ( ), ).pipe(Effect.withSpan("mcp.host.create_server")); + const codeRuntime = config.codeRuntime ?? "sandbox"; + const clientLabel = (): string | undefined => { + const client = server.server.getClientVersion(); + return client ? `${client.name}/${client.version}` : undefined; + }; + const executeWithNativeElicitation = ( code: string, extra: McpRequestJoinKeys, ): Effect.Effect => - engine - .execute(code, { + trackEngineRun( + codeRuntime, + engine.execute(code, { onElicitation: makeMcpElicitationHandler(server, extra.requestId, debugLog), - }) - .pipe(Effect.map(toMcpResult)); + }), + ).pipe(Effect.map(toMcpResult)); const executeCode = ( code: string, @@ -1758,7 +1767,7 @@ export const createExecutorMcpServer = ( if (elicitationMode.mode === "native") { return yield* executeWithNativeElicitation(code, extra); } - const outcome = yield* engine.executeWithPause(code); + const outcome = yield* trackEngineRun(codeRuntime, engine.executeWithPause(code)); debugLog("execute.paused_flow_result", { status: outcome.status, executionId: outcome.status === "paused" ? outcome.execution.id : undefined, @@ -1884,7 +1893,7 @@ export const createExecutorMcpServer = ( if (elicitationMode.mode === "native") { return yield* executeWithNativeElicitation(boundCode, extra); } - const outcome = yield* engine.executeWithPause(boundCode); + const outcome = yield* trackEngineRun(codeRuntime, engine.executeWithPause(boundCode)); debugLog("execute_action.paused_flow_result", { status: outcome.status, executionId: outcome.status === "paused" ? outcome.execution.id : undefined, @@ -2046,8 +2055,8 @@ export const createExecutorMcpServer = ( // --- passthrough call path --- // - // Invoke runs one generated call through the existing execution engine. - // The client approves the generic destructive tool; upstream prompts use + // Invoke calls the tool directly through the execution engine, in the + // host, without a sandbox. The client approves the generic destructive tool; upstream prompts use // native elicitation, or fail with an actionable result when unsupported. const executePassthroughCall = ( address: ToolAddress, @@ -2085,9 +2094,10 @@ export const createExecutorMcpServer = ( unanswerable = ctx.request; return Effect.succeed({ action: "decline" as const }); }; - const outcome = yield* engine.execute(passthroughCallCode(address, args), { - onElicitation, - }); + const outcome = yield* trackEngineRun( + "direct", + engine.call(String(address), args, { onElicitation }), + ); if (unanswerable) return elicitationUnsupportedResult(String(address), unanswerable); return toPassthroughResult(outcome); }).pipe( @@ -2110,7 +2120,11 @@ export const createExecutorMcpServer = ( config.tools, config.connections, config.integrations, - executePassthroughCall, + (address, args, extra) => + logToolCall( + { tool: "invoke", client: clientLabel }, + executePassthroughCall(address, args, extra), + ), config.toolDiscoveryProvider ?? defaultToolDiscoveryProvider, ); } @@ -2123,7 +2137,14 @@ export const createExecutorMcpServer = ( description, inputSchema: { code: z.string().trim().min(1) }, }, - ({ code }, extra) => runToolEffect(executeCode(code, extra), extra), + ({ code }, extra) => + runToolEffect( + logToolCall( + { tool: "execute", client: clientLabel, codeLength: code.length }, + executeCode(code, extra), + ), + extra, + ), ), ).pipe( Effect.withSpan("mcp.host.register_tool", { @@ -2195,14 +2216,17 @@ export const createExecutorMcpServer = ( }, ({ executionId, action, content: rawContent, persist }, extra) => runToolEffect( - resumeExecution( - executionId, - { - action, - content: parseJsonContent(rawContent), - ...(persist === undefined ? {} : { meta: { persist } }), - }, - extra, + logToolCall( + { tool: "resume", client: clientLabel }, + resumeExecution( + executionId, + { + action, + content: parseJsonContent(rawContent), + ...(persist === undefined ? {} : { meta: { persist } }), + }, + extra, + ), ), extra, ), @@ -2222,7 +2246,13 @@ export const createExecutorMcpServer = ( }, }, ({ executionId }, extra) => - runToolEffect(resumeAfterBrowserApproval(executionId, extra), extra), + runToolEffect( + logToolCall( + { tool: "resume", client: clientLabel }, + resumeAfterBrowserApproval(executionId, extra), + ), + extra, + ), ); }).pipe( Effect.withSpan("mcp.host.register_tool", { @@ -2263,9 +2293,13 @@ export const createExecutorMcpServer = ( "Search this integration's tools; empty query lists all. Run results with execute.", inputSchema: { query: z.string().optional() }, }, - ({ query }, extra) => - runToolEffect( - executeCode(searchNamespaceCode(integration, query), extra).pipe( + ({ query }, extra) => { + const code = searchNamespaceCode(integration, query); + return runToolEffect( + logToolCall( + { tool: "search_namespace", client: clientLabel, codeLength: code.length }, + executeCode(code, extra), + ).pipe( Effect.withSpan("mcp.host.tool.namespace_search", { attributes: { "mcp.tool.name": `search_${integration}`, @@ -2274,7 +2308,8 @@ export const createExecutorMcpServer = ( }), ), extra, - ), + ); + }, ); } }).pipe( @@ -2880,7 +2915,13 @@ export const createExecutorMcpServer = ( }, }, ({ code, artifactId }, extra) => - runToolEffect(executeCodeFromApp(code, artifactId, extra), extra), + runToolEffect( + logToolCall( + { tool: "execute_action", client: clientLabel, codeLength: code.length }, + executeCodeFromApp(code, artifactId, extra), + ), + extra, + ), ); executeActionResumeTool = registerAppTool( @@ -2904,10 +2945,13 @@ export const createExecutorMcpServer = ( }, ({ executionId, action, content: rawContent }, extra) => runToolEffect( - resumeExecution( - executionId, - { action, content: parseJsonContent(rawContent) }, - extra, + logToolCall( + { tool: "resume", client: clientLabel }, + resumeExecution( + executionId, + { action, content: parseJsonContent(rawContent) }, + extra, + ), ), extra, ),