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 + 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.ts b/packages/hosts/mcp/src/tool-server.ts index 41a9fa3856..335559465b 100644 --- a/packages/hosts/mcp/src/tool-server.ts +++ b/packages/hosts/mcp/src/tool-server.ts @@ -90,6 +90,7 @@ import { import { MCP_ORG_WRITE_ACCESS_HEADER } from "./seams"; 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()) @@ -209,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 @@ -1655,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 => @@ -1726,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, @@ -1754,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, @@ -1880,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, @@ -2081,7 +2094,10 @@ export const createExecutorMcpServer = ( unanswerable = ctx.request; return Effect.succeed({ action: "decline" as const }); }; - const outcome = yield* engine.call(String(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( @@ -2104,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, ); } @@ -2117,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", { @@ -2189,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, ), @@ -2216,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", { @@ -2257,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}`, @@ -2268,7 +2308,8 @@ export const createExecutorMcpServer = ( }), ), extra, - ), + ); + }, ); } }).pipe( @@ -2874,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( @@ -2898,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, ),