919 lines · 31.2 KB · TypeScript
RawDownload
import { Deferred, Effect, Fiber, Predicate, Queue } from "effect";import * as Cause from "effect/Cause";import * as Exit from "effect/Exit"; import type {  Executor,  InvokeOptions,  ElicitationResponse,  ElicitationHandler,  ElicitationContext,} from "@executor-js/sdk/core";import { CodeExecutionError } from "@executor-js/codemode-core";import type {  CodeExecutor,  ExecuteOutputItem,  ExecuteNotification,  ExecuteResult,  SandboxToolInvoker,} from "@executor-js/codemode-core"; import {  defaultToolDiscoveryProvider,  makeExecutorToolInvoker,  listExecutorSources,  describeTool,  type ToolDiscoveryProvider,} from "./tool-invoker";import { ExecutionToolError } from "./errors";import { buildExecuteDescription } from "./description"; // ---------------------------------------------------------------------------// Types// --------------------------------------------------------------------------- export type ExecutionEngineConfig<E extends Cause.YieldableError = CodeExecutionError> = {  readonly executor: Executor;  readonly codeExecutor: CodeExecutor<E>;  readonly toolDiscoveryProvider?: ToolDiscoveryProvider;}; export type ExecutionResult =  | { readonly status: "completed"; readonly result: ExecuteResult }  | { readonly status: "paused"; readonly execution: PausedExecution }; export type ExecutionCellStatus = "running" | "completed" | "failed" | "terminated"; type ExecutionCellEventPayload =  | {      readonly type: "output";      readonly item: ExecuteOutputItem;    }  | {      readonly type: "yielded";    }  | {      readonly type: "completed";      readonly result: ExecuteResult;    }  | {      readonly type: "failed";      readonly error: string;    }  | {      readonly type: "terminated";    }; export type ExecutionCellEvent = ExecutionCellEventPayload & {  readonly id: number;}; export type ExecutionCellObservation = {  readonly status: ExecutionCellStatus;  readonly cellId: string;  readonly cursor: number;  readonly events: readonly ExecutionCellEvent[];  readonly result?: ExecuteResult;  readonly error?: string;}; export type StartExecutionCellOptions = {  readonly yieldAfterMs?: number;}; export type WaitExecutionCellOptions = {  readonly after?: number;  readonly timeoutMs?: number;}; export type PausedExecution = {  readonly id: string;  readonly elicitationContext: ElicitationContext;}; /** Internal representation with Effect runtime state for pause/resume. */type InternalPausedExecution<E> = PausedExecution & {  readonly response: Deferred.Deferred<typeof ElicitationResponse.Type>;  readonly fiber: Fiber.Fiber<ExecuteResult, E>;  readonly pauseQueue: Queue.Queue<InternalPausedExecution<E>>;}; type InternalExecutionCell<E> = {  readonly id: string;  events: ExecutionCellEvent[];  nextEventId: number;  status: ExecutionCellStatus;  result?: ExecuteResult;  error?: string;  fiber?: Fiber.Fiber<ExecuteResult, E>;  readonly waiters: Set<Deferred.Deferred<void>>;  readonly yieldContinuations: Set<{    readonly resume: () => void;    readonly terminate: () => void;  }>;}; type CellObservationMode = "any-event" | "yield-or-terminal"; const CELL_TERMINATED_MESSAGE = "Execution cell terminated"; export type ResumeResponse = {  readonly action: "accept" | "decline" | "cancel";  readonly content?: Record<string, unknown>;}; // ---------------------------------------------------------------------------// Result formatting// --------------------------------------------------------------------------- const MAX_PREVIEW_CHARS = 30_000; const truncate = (value: string, max: number): string =>  value.length > max    ? `${value.slice(0, max)}\n... [truncated ${value.length - max} chars]`    : value; export const formatExecuteResult = (  result: ExecuteResult,): {  text: string;  structured: Record<string, unknown>;  isError: boolean;} => {  const resultText =    result.result != null      ? typeof result.result === "string"        ? result.result        : JSON.stringify(result.result, null, 2)      : null;   const logText = result.logs && result.logs.length > 0 ? result.logs.join("\n") : null;  const notifications =    result.output      ?.filter(        (          item,        ): item is { readonly type: "notification"; readonly notification: ExecuteNotification } =>          item.type === "notification",      )      .map((item) => item.notification) ?? [];  const notificationField = notifications.length > 0 ? { notifications } : {};   // `emit()` output is shown to the user, not returned to the model, so a  // script that only emits comes back with a null result. Acknowledge the  // emitted items in the envelope so an emit-without-return reads as "output  // went to the user" rather than a silent void.  const emitted = result.output?.length ?? 0;  const emittedNote =    emitted > 0 ? `${emitted} item${emitted === 1 ? "" : "s"} emitted to the user` : null;  const emittedField = emitted > 0 ? { emitted } : {};   if (result.error) {    const parts = [`Error: ${result.error}`, ...(logText ? [`\nLogs:\n${logText}`] : [])];    return {      text: truncate(parts.join("\n"), MAX_PREVIEW_CHARS),      structured: {        status: "error",        error: result.error,        ...emittedField,        ...notificationField,        logs: result.logs ?? [],      },      isError: true,    };  }   const resultPart = resultText    ? truncate(resultText, MAX_PREVIEW_CHARS)    : emittedNote      ? `(no return value; ${emittedNote})`      : "(no result)";  const parts = [resultPart, ...(logText ? [`\nLogs:\n${logText}`] : [])];  return {    text: parts.join("\n"),    structured: {      status: "completed",      result: result.result ?? null,      ...emittedField,      ...notificationField,      logs: result.logs ?? [],    },    isError: false,  };}; export const formatPausedExecution = (  paused: PausedExecution,): {  text: string;  structured: Record<string, unknown>;} => {  const req = paused.elicitationContext.request;  const lines: string[] = [`Execution paused: ${req.message}`];  const isUrlElicitation = Predicate.isTagged(req, "UrlElicitation");  const isFormElicitation = Predicate.isTagged(req, "FormElicitation");  const requestedSchema = isFormElicitation ? req.requestedSchema : undefined;  const hasRequestedSchema =    requestedSchema !== undefined && Object.keys(requestedSchema).length > 0;  const instructions = isUrlElicitation    ? `The user needs to open this URL in a browser and complete the flow. After the user finishes, call the resume tool with executionId "${paused.id}" and action "accept".`    : hasRequestedSchema      ? `Ask the user for values matching requestedSchema. Then call the resume tool with executionId "${paused.id}", action "accept", and content matching requestedSchema. If the user declines, call resume with action "decline" or "cancel".`      : `This is a model-side confirmation gate; there is no browser form to open. Ask the user whether to approve the paused tool call. If the user approves, call the resume tool with executionId "${paused.id}" and action "accept". If the user declines, call resume with action "decline" or "cancel".`;   if (isUrlElicitation) {    lines.push(`\nOpen this URL in a browser:\n${req.url}`);    lines.push('\nAfter the browser flow, call the resume tool with action "accept".');  } else if (hasRequestedSchema) {    lines.push(      "\nAsk the user for a response matching the requested schema, then call the resume tool.",    );    lines.push(`\nRequested schema:\n${JSON.stringify(requestedSchema, null, 2)}`);  } else {    lines.push(      '\nThis is a model-side confirmation gate; no browser form is waiting. Ask the user whether to approve, then call the resume tool with action "accept", "decline", or "cancel".',    );  }   lines.push(`\nexecutionId: ${paused.id}`);  lines.push(`\ninstructions: ${instructions}`);   return {    text: lines.join("\n"),    structured: {      status: "waiting_for_interaction",      executionId: paused.id,      interaction: {        kind: isUrlElicitation ? "url" : "form",        message: req.message,        instructions,        address: String(paused.elicitationContext.address),        args: paused.elicitationContext.args,        ...(isUrlElicitation ? { url: req.url } : {}),        ...(isFormElicitation ? { requestedSchema: req.requestedSchema } : {}),      },    },  };}; // ---------------------------------------------------------------------------// Full invoker (base + discover + describe)// --------------------------------------------------------------------------- const isRecord = (value: unknown): value is Record<string, unknown> =>  typeof value === "object" && value !== null && !Array.isArray(value); const readOptionalLimit = (value: unknown, toolName: string): number | ExecutionToolError => {  if (value === undefined) {    return 12;  }   if (typeof value !== "number" || !Number.isFinite(value) || value <= 0) {    return new ExecutionToolError({      message: `${toolName} limit must be a positive number when provided`,    });  }   return Math.floor(value);}; const readOptionalOffset = (value: unknown, toolName: string): number | ExecutionToolError => {  if (value === undefined) {    return 0;  }   if (typeof value !== "number" || !Number.isFinite(value) || value < 0) {    return new ExecutionToolError({      message: `${toolName} offset must be a non-negative number when provided`,    });  }   return Math.floor(value);}; const makeFullInvoker = (  executor: Executor,  invokeOptions: InvokeOptions,  toolDiscoveryProvider: ToolDiscoveryProvider,): SandboxToolInvoker => {  const base = makeExecutorToolInvoker(executor, { invokeOptions });  return {    invoke: ({ path, args }) => {      if (path === "search") {        if (!isRecord(args)) {          return Effect.fail(            new ExecutionToolError({              message:                "tools.search expects an object: { query?: string; namespace?: string; limit?: number; offset?: number }",            }),          );        }         if (args.query !== undefined && typeof args.query !== "string") {          return Effect.fail(            new ExecutionToolError({              message: "tools.search query must be a string when provided",            }),          );        }         if (args.namespace !== undefined && typeof args.namespace !== "string") {          return Effect.fail(            new ExecutionToolError({              message: "tools.search namespace must be a string when provided",            }),          );        }         const limit = readOptionalLimit(args.limit, "tools.search");        if (Predicate.isTagged(limit, "ExecutionToolError")) {          return Effect.fail(limit);        }         const offset = readOptionalOffset(args.offset, "tools.search");        if (Predicate.isTagged(offset, "ExecutionToolError")) {          return Effect.fail(offset);        }         return toolDiscoveryProvider          .searchTools({            executor,            query: args.query ?? "",            limit,            namespace: args.namespace,            offset,          })          .pipe(            Effect.withSpan("mcp.tool.dispatch", {              attributes: { "mcp.tool.name": path, "executor.tool.builtin": true },            }),          );      }      if (path === "executor.sources.list") {        if (args !== undefined && !isRecord(args)) {          return Effect.fail(            new ExecutionToolError({              message:                "tools.executor.sources.list expects an object: { query?: string; limit?: number; offset?: number }",            }),          );        }         if (isRecord(args) && args.query !== undefined && typeof args.query !== "string") {          return Effect.fail(            new ExecutionToolError({              message: "tools.executor.sources.list query must be a string when provided",            }),          );        }         const limit = readOptionalLimit(          isRecord(args) ? args.limit : undefined,          "tools.executor.sources.list",        );        if (Predicate.isTagged(limit, "ExecutionToolError")) {          return Effect.fail(limit);        }         const offset = readOptionalOffset(          isRecord(args) ? args.offset : undefined,          "tools.executor.sources.list",        );        if (Predicate.isTagged(offset, "ExecutionToolError")) {          return Effect.fail(offset);        }         return listExecutorSources(executor, {          query: isRecord(args) && typeof args.query === "string" ? args.query : undefined,          limit,          offset,        }).pipe(          Effect.withSpan("mcp.tool.dispatch", {            attributes: { "mcp.tool.name": path, "executor.tool.builtin": true },          }),        );      }      if (path === "describe.tool") {        if (!isRecord(args)) {          return Effect.fail(            new ExecutionToolError({              message: "tools.describe.tool expects an object: { path: string }",            }),          );        }         if (typeof args.path !== "string" || args.path.trim().length === 0) {          return Effect.fail(new ExecutionToolError({ message: "describe.tool requires a path" }));        }         if ("includeSchemas" in args) {          return Effect.fail(            new ExecutionToolError({              message: "tools.describe.tool no longer accepts includeSchemas",            }),          );        }         return describeTool(executor, args.path).pipe(          Effect.withSpan("mcp.tool.dispatch", {            attributes: {              "mcp.tool.name": path,              "executor.tool.builtin": true,              "executor.tool.target_path": args.path,            },          }),        );      }      return base.invoke({ path, args });    },  };}; // ---------------------------------------------------------------------------// Execution Engine// --------------------------------------------------------------------------- export type ExecutionEngine<E extends Cause.YieldableError = CodeExecutionError> = {  /**   * Execute code with elicitation handled inline by the provided handler.   * Use this when the host supports elicitation (e.g. MCP with elicitation capability).   *   * Fails with the code executor's typed error `E` (defaults to   * `CodeExecutionError`). Runtimes surface their own `Data.TaggedError`   * subclass, which flows through here unchanged.   */  readonly execute: (    code: string,    options: { readonly onElicitation: ElicitationHandler },  ) => Effect.Effect<ExecuteResult, E>;   /**   * Execute code, intercepting the first elicitation as a pause point.   * Use this when the host doesn't support inline elicitation.   * Returns either a completed result or a paused execution that can be resumed.   */  readonly executeWithPause: (code: string) => Effect.Effect<ExecutionResult, E>;   /**   * Resume a paused execution. Returns a completed result, a new pause, or   * null if the executionId was not found.   */  readonly resume: (    executionId: string,    response: ResumeResponse,  ) => Effect.Effect<ExecutionResult | null, E>;   /**   * Inspect a paused execution without resuming it. Returns null if the id is   * unknown or has already been resumed.   */  readonly getPausedExecution: (executionId: string) => Effect.Effect<PausedExecution | null>;   /**   * Start a persistent execution cell. The cell keeps running after the first   * observation when it yields or exceeds `yieldAfterMs`.   */  readonly startCell: (    code: string,    options?: StartExecutionCellOptions,  ) => Effect.Effect<ExecutionCellObservation, E>;   /**   * Wait for new cell events after `after`. Returns null when the cell id is   * unknown.   */  readonly waitCell: (    cellId: string,    options?: WaitExecutionCellOptions,  ) => Effect.Effect<ExecutionCellObservation | null, E>;   /**   * Stop a running cell and return its current observation. Returns null when   * the cell id is unknown.   */  readonly terminateCell: (cellId: string) => Effect.Effect<ExecutionCellObservation | null>;   /**   * Get the dynamic tool description (workflow + namespaces).   */  readonly getDescription: Effect.Effect<string>;};

Lines 1–500 of 919.