import { Duration, Effect, Match, Option, Schema } from "effect";import * as Cause from "effect/Cause";import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";import { ContentBlockSchema, type ContentBlock } from "@modelcontextprotocol/sdk/types.js";import type {  jsonSchemaValidator,  JsonSchemaType,  JsonSchemaValidator,} from "@modelcontextprotocol/sdk/validation/types.js";import { Validator } from "@cfworker/json-schema";import * as z from "zod/v4"; import { isToolFile } from "@executor-js/sdk";import type {  ElicitationResponse,  ElicitationHandler,  ElicitationContext,  ElicitationRequest,  ToolFileValue,} from "@executor-js/sdk";import type * as Tracer from "effect/Tracer";import {  createExecutionEngine,  formatExecuteResult,  formatPausedExecution,  formatTtlDuration,  findSkill,  renderSkillsIndex,  EXECUTE_SKILL,  INTEGRATION_INVENTORY_HEADER,  type ExecutionEngine,  type ExecutionEngineConfig,  type ResumeResponse,  type ExecutionResult,  type PausedExecution,  type PausedExecutionDeadline,} from "@executor-js/execution"; // ---------------------------------------------------------------------------// Workers-compatible JSON Schema validator (replaces Ajv which uses new Function())// --------------------------------------------------------------------------- class CfWorkerJsonSchemaValidator implements jsonSchemaValidator {  getValidator<T>(schema: JsonSchemaType): JsonSchemaValidator<T> {    const validator = new Validator(schema as Record<string, unknown>, "2020-12", false);    return (input: unknown) => {      const result = validator.validate(input);      if (result.valid) {        return { valid: true, data: input as T, errorMessage: undefined };      }      const errorMessage = result.errors.map((e) => `${e.instanceLocation}: ${e.error}`).join("; ");      return { valid: false, data: undefined, errorMessage };    };  }} // ---------------------------------------------------------------------------// Config// --------------------------------------------------------------------------- type SharedMcpServerConfig = {  /**   * Pre-built `execute` tool description. When provided, the factory skips   * its internal `engine.getDescription` yield. Useful when the caller   * wants to compute the description inside its own Effect tracer context   * so sub-spans (`executor.integrations.list`, `executor.tools.list`) nest as   * children of the caller's root span.   */  readonly description?: string;  /**   * Parent span override for engine calls. The factory captures the   * caller's context at construction time, but `Effect.runPromiseWith`   * starts a fresh fiber per SDK callback — so the `currentSpan`   * FiberRef resets to root unless explicitly anchored.   *   * Accepts either a fixed span (per-request McpServer instances) or a   * getter (session-scoped instances that need to anchor each callback   * under whichever request triggered it; see the Cloud DO).   */  readonly parentSpan?: Tracer.AnySpan | (() => Tracer.AnySpan | undefined);  /**   * Enable verbose MCP capability / elicitation debug logging.   */  readonly debug?: boolean;  /**   * Controls how elicitation is handled for this MCP connection. The default   * is model-managed resume, where paused executions expose interaction   * metadata and the model can call `resume` with the user's response.   */  readonly elicitationMode?:    | {        readonly mode: "browser";        readonly approvalUrl: (executionId: string) => string;      }    | {        readonly mode: "model";      }    | {        readonly mode: "native";      };  readonly browserApprovalStore?: BrowserApprovalStore;  /**   * Host-owned lifecycle for paused executions. The MCP server reports pause   * boundaries; the host decides whether that means a keepAlive lease, browser   * wait, durable record, or no-op.   */  readonly pausedExecutionHooks?: PausedExecutionHooks;  /**   * Host-provided approval lease duration. When present, paused payloads carry   * an absolute deadline and hooks receive the same deadline.   */  readonly pausedExecutionLeaseMs?: number;  /**   * Optional host-owned model resume fallback. Used by Cloudflare session   * Durable Objects to route a resume miss to the session that owns the pause.   */  readonly resumeFallback?: (    executionId: string,    response: ResumeResponse,  ) => Effect.Effect<ResumeFallbackOutcome | null, unknown>;}; export type ExecutorMcpServerConfig<E extends Cause.YieldableError = Cause.YieldableError> =  | (ExecutionEngineConfig<E> & SharedMcpServerConfig)  | ({ readonly engine: ExecutionEngine<E> } & SharedMcpServerConfig)  | (ExecutionEngineConfig<E> & SharedMcpServerConfig & { readonly stateless: true })  | ({ readonly engine: ExecutionEngine<E>; readonly stateless: true } & SharedMcpServerConfig); export type BrowserApprovalStore = {  readonly takeResponse: (executionId: string) => Effect.Effect<ResumeResponse | null>;  readonly waitForResponse?: (executionId: string) => Effect.Effect<ResumeResponse | null>;}; export const PAUSED_APPROVAL_TIMEOUT_MS = 4 * 60 * 1000;const BROWSER_APPROVAL_WAIT_TIMEOUT_MS = PAUSED_APPROVAL_TIMEOUT_MS + 1000; export type PausedExecutionHooks = {  readonly onExecutionPaused?: (    executionId: string,    deadline: PausedExecutionDeadline | undefined,  ) => Effect.Effect<void>;  readonly onResumeStarted?: (executionId: string) => Effect.Effect<void>;  readonly onResumeSettled?: (executionId: string) => Effect.Effect<void>;}; export type ResumeUnavailableStatus =  | "execution_not_found"  | "execution_expired"  | "execution_forbidden"  | "execution_already_settled"; export type ResumeFallbackOutcome =  | {      readonly status: "result";      readonly result: McpToolResult;    }  | {      readonly status: Exclude<ResumeUnavailableStatus, "execution_not_found">;      readonly ttlMs?: number;    }  | {      readonly status: "execution_not_found";    }; // ---------------------------------------------------------------------------// Elicitation bridge// --------------------------------------------------------------------------- const getElicitationSupport = (server: McpServer): { form: boolean; url: boolean } => {  const capabilities = server.server.getClientCapabilities();  if (capabilities === undefined || !capabilities.elicitation) return { form: false, url: false };  const elicitation = capabilities.elicitation as Record<string, unknown>;  return { form: Boolean(elicitation.form), url: Boolean(elicitation.url) };}; const readDebugDefault = (): boolean => {  if (typeof process === "undefined" || !process.env) return false;  const value = process.env.EXECUTOR_MCP_DEBUG;  return value === "1" || value === "true";}; const capabilitySnapshot = (server: McpServer) => ({  clientCapabilities: server.server.getClientCapabilities() ?? null,  elicitationSupport: getElicitationSupport(server),}); type ElicitInputParams =  | {      mode?: "form";      message: string;      requestedSchema: { readonly [key: string]: unknown };    }  | { mode: "url"; message: string; url: string; elicitationId: string }; const elicitationRequestTag = (request: ElicitationRequest): ElicitationRequest["_tag"] =>  Match.value(request).pipe(    Match.tag("UrlElicitation", () => "UrlElicitation" as const),    Match.tag("FormElicitation", () => "FormElicitation" as const),    Match.exhaustive,  ); const requestedSchemaIsNonEmpty = (request: ElicitationRequest): boolean =>  Match.value(request).pipe(    Match.tag("FormElicitation", (req) => Object.keys(req.requestedSchema).length > 0),    Match.tag("UrlElicitation", () => false),    Match.exhaustive,  ); const elicitationRequestUrl = (request: ElicitationRequest): string | undefined =>  Match.value(request).pipe(    Match.tag("UrlElicitation", (req): string | undefined => req.url),    Match.tag("FormElicitation", (): string | undefined => undefined),    Match.exhaustive,  ); const pausedInteractionKind = (request: ElicitationRequest): ElicitationRequest["_tag"] =>  elicitationRequestTag(request); const elicitationRequestToParams: (request: ElicitationRequest) => ElicitInputParams =  Match.type<ElicitationRequest>().pipe(    Match.tag("UrlElicitation", (req) => ({      mode: "url" as const,      message: req.message,      url: req.url,      elicitationId: req.elicitationId,    })),    Match.tag("FormElicitation", (req) => ({      message: req.message,      // The MCP SDK validates requestedSchema as a JSON Schema with      // `type: "object"` and `properties`. For approval-only elicitations      // where no fields are needed, provide a minimal valid schema.      requestedSchema:        Object.keys(req.requestedSchema).length === 0          ? { type: "object" as const, properties: {} }          : req.requestedSchema,    })),    Match.exhaustive,  ); const makeMcpElicitationHandler =  (    server: McpServer,    debugLog?: (event: string, data: Record<string, unknown>) => void,  ): ElicitationHandler =>  (ctx: ElicitationContext): Effect.Effect<typeof ElicitationResponse.Type> => {    const { url: supportsUrl } = getElicitationSupport(server);     // If client doesn't support url mode, fall back to a form asking the user    // to visit the URL manually and confirm when done.    const params = Match.value(ctx.request).pipe(      Match.tag(        "UrlElicitation",        (req): ElicitInputParams =>          !supportsUrl            ? {                message: `${req.message}\n\nPlease visit this URL:\n${req.url}\n\nClick accept once you have completed the flow.`,                requestedSchema: { type: "object" as const, properties: {} },              }            : elicitationRequestToParams(req),      ),      Match.tag("FormElicitation", (req): ElicitInputParams => elicitationRequestToParams(req)),      Match.exhaustive,    );     return Effect.promise(async (): Promise<typeof ElicitationResponse.Type> => {      const requestTag = elicitationRequestTag(ctx.request);      debugLog?.("elicitation.request", {        requestTag,        supportsUrl,        message: ctx.request.message,        hasRequestedSchema: requestedSchemaIsNonEmpty(ctx.request),        url: elicitationRequestUrl(ctx.request),        clientCapabilities: server.server.getClientCapabilities() ?? null,      });       // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: MCP SDK elicitInput is a Promise API; failures become a cancel response      try {        const response = await server.server.elicitInput(          params as Parameters<typeof server.server.elicitInput>[0],        );         debugLog?.("elicitation.response", {          requestTag,          action: response.action,          hasContent:            typeof response.content === "object" &&            response.content !== null &&            Object.keys(response.content).length > 0,        });         return {          action: response.action as typeof ElicitationResponse.Type.action,          content: response.content,        };      } catch (err) {        const error = formatBoundaryError(err);        debugLog?.("elicitation.error", {          requestTag,          error,          clientCapabilities: server.server.getClientCapabilities() ?? null,        });        return { action: "cancel" as const } as ElicitationResponse;      }    });  }; const formatBoundaryError = (err: unknown): { name?: string; message: string; stack?: string } => {  // oxlint-disable-next-line executor/no-instanceof-error, executor/no-unknown-error-message -- boundary: SDK Promise rejection supplies unknown JS errors for logging only  if (err instanceof Error) return { name: err.name, message: err.message, stack: err.stack };  // oxlint-disable-next-line executor/no-unknown-error-message -- boundary: fallback log formatting for unknown SDK Promise rejection values  return { message: String(err) };}; // ---------------------------------------------------------------------------// MCP result formatting// --------------------------------------------------------------------------- export type McpToolResult = {  content: ContentBlock[];  structuredContent?: Record<string, unknown>;  isError?: boolean;}; type FormattedExecuteInput = Parameters<typeof formatExecuteResult>[0];type ExecuteOutputItem = NonNullable<FormattedExecuteInput["output"]>[number]; const TEXT_FILE_CONTENT_MAX_CHARS = 64_000; const isRecord = (value: unknown): value is Record<string, unknown> =>  typeof value === "object" && value !== null && !Array.isArray(value); const toolFileName = (file: ToolFileValue): string => file.name ?? "tool-output"; const fileResourceUri = (file: ToolFileValue): string =>  `executor-file:///${encodeURIComponent(toolFileName(file))}`; const normalizedMimeType = (file: ToolFileValue): string =>  file.mimeType.split(";")[0]?.trim().toLowerCase() ?? ""; const toolFileKind = (file: ToolFileValue): "image" | "audio" | "text" | "resource" => {  const mimeType = normalizedMimeType(file);  if (mimeType.startsWith("image/")) return "image";  if (mimeType.startsWith("audio/")) return "audio";  if (    mimeType.startsWith("text/") ||    mimeType === "application/json" ||    mimeType.endsWith("+json") ||    mimeType === "application/xml" ||    mimeType.endsWith("+xml") ||    mimeType === "application/javascript" ||    mimeType === "application/x-javascript" ||    mimeType === "application/yaml" ||    mimeType === "application/x-yaml"  ) {    return "text";  }  return "resource";}; const bytesFromBase64 = (base64: string): Uint8Array => {  const binary = atob(base64);  const bytes = new Uint8Array(binary.length);  for (let index = 0; index < binary.length; index += 1) {    bytes[index] = binary.charCodeAt(index);  }  return bytes;}; const decodeTextFile = (file: ToolFileValue): string => {  const text = new TextDecoder("utf-8", { fatal: false }).decode(bytesFromBase64(file.data));  if (text.length <= TEXT_FILE_CONTENT_MAX_CHARS) return text;  return `${text.slice(0, TEXT_FILE_CONTENT_MAX_CHARS)}\n\n[truncated ${    text.length - TEXT_FILE_CONTENT_MAX_CHARS  } characters]`;}; const toolFileContent = (file: ToolFileValue): ContentBlock[] => {  const kind = toolFileKind(file);  if (kind === "image") {    return [{ type: "image", data: file.data, mimeType: file.mimeType }];  }  if (kind === "audio") {    return [{ type: "audio", data: file.data, mimeType: file.mimeType }];  }  if (kind === "text") {    return [{ type: "text", text: decodeTextFile(file) }];  }  return [    {      type: "resource",      resource: {        uri: fileResourceUri(file),        mimeType: file.mimeType,        blob: file.data,      },    },  ];}; const toolFileSummaryLine = (file: ToolFileValue, index?: number): string => {  const prefix = index === undefined ? "" : `${index + 1}. `;  return `${prefix}${toolFileName(file)} (${file.mimeType}, ${file.byteLength} bytes)`;}; const outputFileContent = (file: ToolFileValue): ContentBlock[] => [  {    type: "text",    text: `File output: ${toolFileSummaryLine(file)}`,  },  ...toolFileContent(file),]; const isFileOutputItem = (  item: ExecuteOutputItem,): item is { readonly type: "file"; readonly file: ToolFileValue } =>  isRecord(item) && item.type === "file" && isToolFile(item.file); const isMcpContentBlock = (value: unknown): value is ContentBlock =>  ContentBlockSchema.safeParse(value).success; const isContentOutputItem = (  item: ExecuteOutputItem,): item is { readonly type: "content"; readonly content: ContentBlock } =>  isRecord(item) && item.type === "content" && isMcpContentBlock(item.content); const outputItemContent = (item: ExecuteOutputItem): ContentBlock[] => {  if (isFileOutputItem(item)) {    return outputFileContent(item.file);  }  if (isContentOutputItem(item)) {    return [item.content];  }  return [{ type: "text", text: "Invalid execution output item omitted." }];}; const toMcpOutputResult = (  result: FormattedExecuteInput,  output: readonly ExecuteOutputItem[],): McpToolResult => {  const formatted = formatExecuteResult(result);  const content = output.flatMap(outputItemContent);  const extraText: string[] = [];  if (result.error) {    extraText.push(formatted.text);  } else if (result.logs && result.logs.length > 0) {    extraText.push(`Logs:\n${result.logs.join("\n")}`);  }  content.push(...extraText.map((text): ContentBlock => ({ type: "text", text })));   return {    content,    structuredContent: formatted.structured,    isError: formatted.isError || undefined,  };}; const toMcpResult = (result: FormattedExecuteInput): McpToolResult => {  if (result.output && result.output.length > 0) return toMcpOutputResult(result, result.output);  const formatted = formatExecuteResult(result);  return {    content: [{ type: "text", text: formatted.text }],    structuredContent: formatted.structured,    isError: formatted.isError || undefined,  };}; const toMcpPausedResult = (formatted: ReturnType<typeof formatPausedExecution>): McpToolResult => ({  content: [{ type: "text", text: formatted.text }],  structuredContent: formatted.structured,}); export const formatMcpExecutionOutcome = (  outcome: ExecutionResult,  options?: { readonly pausedDeadline?: PausedExecutionDeadline },): McpToolResult =>  outcome.status === "completed"    ? toMcpResult(outcome.result)    : toMcpPausedResult(        formatPausedExecution(outcome.execution, { deadline: options?.pausedDeadline }),      ); // `execute` failures reaching the MCP host are infra defects — domain// failures from tools are now expressed as `ToolResult` values (success// channel) and flow through `formatExecuteResult`. Emit an opaque// generic plus a fresh correlation id and log the cause out-of-band so// the model can't read internal context off `.message`.const newCorrelationId = (): string =>  Math.floor(Math.random() * 0x1_0000_0000)    .toString(16)    .padStart(8, "0"); const defaultResumeApprovalUrl = (executionId: string): string =>  `/resume/${encodeURIComponent(executionId)}`; const browserApprovalReturnPrompt =  "Return text to the user telling them to approve the action at this approvalUrl. Only after you have prompted the user, call the `resume` tool with this executionId; `resume` will wait for the user's browser decision."; const formatResumeApprovalRequired = (input: {  readonly executionId: string;  readonly approvalUrl: string;

Lines 1–500 of 1100. Load more