diff --git a/README.md b/README.md index 584204d7..a006d620 100644 --- a/README.md +++ b/README.md @@ -56,6 +56,7 @@ cd my-app && npm run dev - ๐Ÿงช **Testing utilities** โ€” Property-based testing, UI mocks, and LLM evaluation - ๐Ÿ“ฆ **Flexible deployment** โ€” Express server, serverless, or stdio - ๐Ÿ”€ **API versioning** โ€” Expose multiple versions from a single app +- ๐Ÿ”„ **Workflow engine** โ€” Compose multi-step workflows with parallel execution, branching, and retry policies ## Table of Contents @@ -194,6 +195,74 @@ export function GreetingWidget() { For complete examples including API versioning, React component UIs, and advanced patterns, see the [Examples](#examples) section. +### Workflow Engine + +Compose multi-step workflows as MCP tools with parallel execution, conditional branching, and retry policies: + +```typescript +import { createApp, workflow, toolStep, customStep } from "@mcp-apps-kit/core"; +import { z } from "zod"; + +const orderWorkflow = workflow("process_order") + .describe("Process a customer order end-to-end") + .input({ + orderId: z.string(), + customerId: z.string(), + }) + .output({ + success: z.boolean(), + receiptId: z.string().optional(), + }) + // Sequential steps + .step("validate", toolStep("validate_order")) + .step("payment", toolStep("process_payment"), { + retry: { maxAttempts: 3, delay: 1000, backoff: "exponential" }, + }) + // Parallel execution + .parallel("notify", [toolStep("send_email"), toolStep("send_sms")]) + // Conditional branching + .branch("shipping", { + when: (ctx) => ctx.outputs.validate.isDigital, + then: [customStep(async () => ({ delivered: true }))], + else: [toolStep("create_shipment")], + }) + .build(); + +const app = createApp({ + name: "order-service", + version: "1.0.0", + tools: { + validate_order: defineTool({ + /* ... */ + }), + process_payment: defineTool({ + /* ... */ + }), + send_email: defineTool({ + /* ... */ + }), + send_sms: defineTool({ + /* ... */ + }), + create_shipment: defineTool({ + /* ... */ + }), + // Register the workflow as a tool + process_order: orderWorkflow, + }, +}); +``` + +Features include: + +- **Sequential steps** with `step()` for ordered execution +- **Parallel execution** with `parallel()` for concurrent operations +- **Conditional branching** with `branch()` for dynamic flow control +- **Retry policies** with exponential/linear backoff +- **Timeout handling** per step +- **External MCP calls** to other MCP servers with `externalStep()` +- **Production-ready** lifecycle management for servers and edge functions + ## Deployment ### Express (default) diff --git a/examples/minimal/src/index.ts b/examples/minimal/src/index.ts index 09835073..355979ef 100644 --- a/examples/minimal/src/index.ts +++ b/examples/minimal/src/index.ts @@ -12,6 +12,9 @@ import { createApp, defineTool, tool, + workflow, + toolStep, + customStep, type ClientToolsFromCore, iconFromFile, } from "@mcp-apps-kit/core"; @@ -19,6 +22,8 @@ import { defineReactUI } from "@mcp-apps-kit/ui-react-builder"; import { GreetingWidgetV1 } from "./ui/GreetingWidgetV1"; import { GreetingWidgetV2 } from "./ui/GreetingWidgetV2"; import { EchoWidget } from "./ui/EchoWidget"; +import { WorkflowWidget } from "./ui/WorkflowWidget"; +import { AdvancedWorkflowWidget } from "./ui/AdvancedWorkflowWidget"; import { z } from "zod"; // ============================================================================= @@ -170,6 +175,195 @@ const echoToolV3 = defineTool({ }, }); +// ============================================================================= +// V4: Workflow Engine Demo - Multi-Step Tool Composition +// ============================================================================= + +/** + * Demonstrates the workflow engine feature. + * Workflows allow you to compose multi-step tools from existing tools and custom logic. + * + * This example creates a "greet_and_echo" workflow that: + * 1. Greets a person using the greet tool + * 2. Transforms the greeting message + * 3. Echoes the transformed message using the echo tool + * 4. Combines both results with a timestamp + */ + +// First, define the individual tools that the workflow will use +const greetForWorkflowTool = defineTool({ + title: "Greet For Workflow", + description: "Greet someone (internal tool for workflow)", + input: { + name: z.string().describe("Name to greet"), + }, + output: { + message: z.string(), + }, + visibility: "model", // Only visible to AI model, not in app UI + handler: async (input) => { + return { + message: `Hello, ${input.name}!`, + }; + }, +}); + +const echoForWorkflowTool = defineTool({ + title: "Echo For Workflow", + description: "Echo a message (internal tool for workflow)", + input: { + message: z.string().describe("Message to echo"), + uppercase: z.boolean().optional().describe("Convert to uppercase"), + }, + output: { + echo: z.string(), + }, + visibility: "model", // Only visible to AI model, not in app UI + handler: async (input) => { + const echo = input.uppercase ? input.message.toUpperCase() : input.message; + return { echo }; + }, +}); + +// Now create a workflow that composes these tools +const greetAndEchoWorkflow = workflow("greet_and_echo") + .describe("Greet someone and echo their greeting with a fun twist") + .input({ + name: z.string().describe("Person's name to greet"), + excitement: z.number().min(1).max(10).default(5).describe("Excitement level (1-10)"), + }) + .output({ + greet_and_echo: z.object({ + greeting: z.string(), + echo: z.string(), + excitementLevel: z.number(), + timestamp: z.string(), + }), + }) + // Step 1: Greet the person + .step("greet", toolStep("greet_for_workflow"), { + mapInput: (ctx) => ({ + name: (ctx.input as { name: string }).name, + }), + }) + // Step 2: Add custom excitement transformation + .step( + "add_excitement", + customStep(async (ctx) => { + const greetingMsg = (ctx.outputs.greet as { message: string }).message; + const excitement = (ctx.input as { excitement: number }).excitement; + const exclamations = "!".repeat(excitement); + return { + enhancedMessage: `${greetingMsg}${exclamations}`, + }; + }) + ) + // Step 3: Echo the enhanced message in uppercase + .step("echo", toolStep("echo_for_workflow"), { + mapInput: (ctx) => ({ + message: (ctx.outputs.add_excitement as { enhancedMessage: string }).enhancedMessage, + uppercase: true, + }), + }) + // Step 4: Combine results + .step( + "combine", + customStep(async (ctx) => { + const greeting = (ctx.outputs.greet as { message: string }).message; + const echo = (ctx.outputs.echo as { echo: string }).echo; + const excitement = (ctx.input as { excitement: number }).excitement; + + return { + greet_and_echo: { + greeting, + echo, + excitementLevel: excitement, + timestamp: new Date().toISOString(), + }, + }; + }) + ) + // Add interactive UI for the workflow + .ui( + defineReactUI({ + component: WorkflowWidget, + name: "Workflow Engine Widget", + description: "Interactive UI demonstrating multi-step workflow execution", + prefersBorder: true, + }) + ) + .build(); + +// Example of a workflow with parallel execution and branching +const advancedWorkflow = workflow("process_greeting") + .describe("Advanced workflow with parallel steps and conditional logic") + .input({ + names: z.array(z.string()).describe("List of names to greet"), + format: z.enum(["formal", "casual"]).default("casual").describe("Greeting format"), + }) + .output({ + summary: z.string(), + greetings: z.array(z.string()), + format: z.string(), + }) + // Parallel step: Greet all names simultaneously + .parallel("greet_all", [ + customStep(async (ctx) => { + const names = (ctx.input as { names: string[] }).names; + return { count: names.length }; + }), + customStep(async (ctx) => { + const names = (ctx.input as { names: string[] }).names; + return { longestName: names.reduce((a, b) => (a.length > b.length ? a : b), "") }; + }), + ]) + // Conditional branching based on format + .branch("format_greeting", { + when: (ctx) => (ctx.input as { format: string }).format === "formal", + then: [ + customStep(async (ctx) => ({ + prefix: "Dear", + suffix: "Sincerely yours", + })), + ], + else: [ + customStep(async (ctx) => ({ + prefix: "Hey", + suffix: "Cheers", + })), + ], + }) + // Final step: combine everything + .step( + "finalize", + customStep(async (ctx) => { + const names = (ctx.input as { names: string[] }).names; + const format = (ctx.input as { format: string }).format; + const parallelResults = ctx.outputs.greet_all as [{ count: number }, { longestName: string }]; + const branchResults = ctx.outputs.format_greeting as [{ prefix: string; suffix: string }]; + + const formatData = branchResults[0]; + const greetings = names.map((name) => `${formatData?.prefix} ${name}`); + const summary = `Processed ${parallelResults[0].count} greetings in ${format} format`; + + return { + summary, + greetings, + format, + }; + }) + ) + // Add interactive UI for the advanced workflow + .ui( + defineReactUI({ + component: AdvancedWorkflowWidget, + name: "Advanced Workflow Widget", + description: "Interactive UI demonstrating parallel execution and conditional branching", + prefersBorder: true, + }) + ) + .build(); + // ============================================================================= // Create Versioned App // ============================================================================= @@ -229,6 +423,19 @@ const app = createApp({ }, // v3 demonstrates inline schema syntax }, + v4: { + version: "4.0.0", + tools: { + // Internal tools used by workflows (visibility: "model") + greet_for_workflow: greetForWorkflowTool, + echo_for_workflow: echoForWorkflowTool, + + // Workflows exposed as tools + greet_and_echo: greetAndEchoWorkflow, + process_greeting: advancedWorkflow, + }, + // v4 demonstrates the workflow engine + }, }, }); @@ -251,12 +458,18 @@ Endpoints: - v2 MCP: http://localhost:${port}/v2/mcp (uses API transport for logging) - v2 Logs: http://localhost:${port}/api/logs (debug log API endpoint) - v3 MCP: http://localhost:${port}/v3/mcp (inline schema syntax demo) + - v4 MCP: http://localhost:${port}/v4/mcp (workflow engine demo) - Health: http://localhost:${port}/health Debug logging: - v1: Uses log_debug MCP tool (default for MCP adapter) - v2: Uses HTTP API transport at /api/logs (ideal for OpenAI/ChatGPT) - v3: Default MCP logging (demonstrates inline schema syntax) + - v4: Default MCP logging (demonstrates workflow engine) + +Try the workflow tools: + - greet_and_echo: Compose greet + custom logic + echo + - process_greeting: Parallel execution + conditional branching `); }); } @@ -284,3 +497,12 @@ export type GreetOutputV2 = z.infer; // V3 types (inline schema syntax - types inferred directly from tool definition) export type AppToolsV3 = { echo: typeof echoToolV3 }; export type AppClientToolsV3 = ClientToolsFromCore; + +// V4 types (workflow engine) +export type AppToolsV4 = { + greet_for_workflow: typeof greetForWorkflowTool; + echo_for_workflow: typeof echoForWorkflowTool; + greet_and_echo: typeof greetAndEchoWorkflow; + process_greeting: typeof advancedWorkflow; +}; +export type AppClientToolsV4 = ClientToolsFromCore; diff --git a/examples/minimal/src/ui/AdvancedWorkflowWidget.tsx b/examples/minimal/src/ui/AdvancedWorkflowWidget.tsx new file mode 100644 index 00000000..303514bb --- /dev/null +++ b/examples/minimal/src/ui/AdvancedWorkflowWidget.tsx @@ -0,0 +1,248 @@ +/** + * Advanced Workflow Widget Component + * + * A React component for displaying advanced workflow results with parallel + * execution and conditional branching. + */ + +import { useEffect, useState } from "react"; +import { useToolResult, useHostContext, useAppsClient } from "@mcp-apps-kit/ui-react"; +import type { AppClientToolsV4 } from "../index"; + +export function AdvancedWorkflowWidget() { + const result = useToolResult(); + const { theme } = useHostContext(); + const client = useAppsClient(); + + const [isModalOpen, setIsModalOpen] = useState(false); + const [workflowInput, setWorkflowInput] = useState({ + names: ["Alice", "Bob"], + format: "casual" as "formal" | "casual", + }); + const [isLoading, setIsLoading] = useState(false); + const [workflowResult, setWorkflowResult] = useState<{ + summary?: string; + greetings?: string[]; + format?: string; + } | null>(null); + const [errorMessage, setErrorMessage] = useState(null); + + // Handle both wrapped and unwrapped result formats + const rawResult = result?.process_greeting ?? result; + const output = + workflowResult ?? + (rawResult && + typeof rawResult === "object" && + "summary" in rawResult && + "greetings" in rawResult && + "format" in rawResult + ? (rawResult as { + summary: string; + greetings: string[]; + format: string; + }) + : undefined); + + useEffect(() => { + if (typeof document !== "undefined") { + document.documentElement.className = theme; + } + }, [theme]); + + const handleRunWorkflow = async () => { + if (workflowInput.names.length === 0) return; + + setIsLoading(true); + setErrorMessage(null); + + try { + const response = await client.tools.callProcess_greeting({ + names: workflowInput.names, + format: workflowInput.format, + }); + + // Extract from structuredContent if present + const result = + (response as { structuredContent?: typeof response }).structuredContent ?? response; + + setWorkflowResult(result); + setIsModalOpen(false); + } catch (error) { + const msg = error instanceof Error ? error.message : String(error); + console.error("Workflow failed:", msg); + setErrorMessage(msg); + } finally { + setIsLoading(false); + } + }; + + const handleAddName = () => { + setWorkflowInput({ + ...workflowInput, + names: [...workflowInput.names, ""], + }); + }; + + const handleRemoveName = (index: number) => { + setWorkflowInput({ + ...workflowInput, + names: workflowInput.names.filter((_, i) => i !== index), + }); + }; + + const handleUpdateName = (index: number, value: string) => { + const newNames = [...workflowInput.names]; + newNames[index] = value; + setWorkflowInput({ ...workflowInput, names: newNames }); + }; + + return ( +
+
Advanced Workflow (v4)
+ + {output && output.summary ? ( +
+

๐Ÿš€ Advanced Workflow Complete!

+ +
+
+

Summary

+

{output.summary}

+
+ + {output.greetings && output.greetings.length > 0 && ( +
+

Greetings ({output.format})

+
    + {output.greetings.map((greeting, idx) => ( +
  • {greeting}
    • + ))} +
+
+ )} + +
+

Features Demonstrated

+
    +
  • โœ… Parallel Execution
    • +
  • โœ… Conditional Branching
    • +
  • โœ… Custom Step Logic
    • +
+
+
+ + +
+ ) : ( +
+

๐ŸŽฏ Advanced Workflow Engine

+

This workflow demonstrates advanced features:

+
    +
  • + Parallel execution: Process multiple names simultaneously +
    • +
  • + Conditional branching: Formal vs casual greeting format +
    • +
  • + Custom logic: Transform and combine results +
    • +
+ +
+ )} + + {isModalOpen && ( +
setIsModalOpen(false)}> +
e.stopPropagation()}> +

Configure Advanced Workflow

+ +
+ + {workflowInput.names.map((name, index) => ( +
+ handleUpdateName(index, e.target.value)} + placeholder={`Name ${index + 1}`} + /> + {workflowInput.names.length > 1 && ( + + )} +
+ ))} + +
+ +
+ +
+ + +
+
+ + {errorMessage &&

{errorMessage}

} + +
+ + +
+
+
+ )} +
+ ); +} + +export default AdvancedWorkflowWidget; diff --git a/examples/minimal/src/ui/WorkflowWidget.tsx b/examples/minimal/src/ui/WorkflowWidget.tsx new file mode 100644 index 00000000..358c50b1 --- /dev/null +++ b/examples/minimal/src/ui/WorkflowWidget.tsx @@ -0,0 +1,207 @@ +/** + * Workflow Widget Component + * + * A React component for displaying workflow execution results. + * Demonstrates UI integration with the workflow engine. + */ + +import { useEffect, useState } from "react"; +import { useToolResult, useHostContext, useAppsClient } from "@mcp-apps-kit/ui-react"; +import type { AppClientToolsV4 } from "../index"; + +export function WorkflowWidget() { + const result = useToolResult(); + const { theme } = useHostContext(); + const client = useAppsClient(); + + const [isModalOpen, setIsModalOpen] = useState(false); + const [workflowInput, setWorkflowInput] = useState({ + name: "", + excitement: 5, + }); + const [isLoading, setIsLoading] = useState(false); + const [workflowResult, setWorkflowResult] = useState<{ + greeting: string; + echo: string; + excitementLevel: number; + timestamp: string; + } | null>(null); + const [errorMessage, setErrorMessage] = useState(null); + + // Handle both wrapped and unwrapped result formats + // The result structure is: { greet_and_echo: { greeting, echo, excitementLevel, timestamp } } + const rawResult = result?.greet_and_echo ?? result; + const extractedData = + rawResult && typeof rawResult === "object" && "greet_and_echo" in rawResult + ? (rawResult.greet_and_echo as { + greeting: string; + echo: string; + excitementLevel: number; + timestamp: string; + }) + : rawResult && typeof rawResult === "object" && "greeting" in rawResult + ? (rawResult as { + greeting: string; + echo: string; + excitementLevel: number; + timestamp: string; + }) + : undefined; + + const output = workflowResult ?? extractedData; + + useEffect(() => { + if (typeof document !== "undefined") { + document.documentElement.className = theme; + } + }, [theme]); + + const handleRunWorkflow = async () => { + if (!workflowInput.name.trim()) return; + + setIsLoading(true); + setErrorMessage(null); + + try { + const response = await client.tools.callGreet_and_echo({ + name: workflowInput.name.trim(), + excitement: workflowInput.excitement, + }); + + // Extract from structuredContent if present + const result = + (response as { structuredContent?: typeof response }).structuredContent ?? response; + + // Handle nested greet_and_echo structure + // The workflow returns: { greet_and_echo: { greeting, echo, excitementLevel, timestamp } } + const finalResult = + result && typeof result === "object" && "greet_and_echo" in result + ? result.greet_and_echo + : result; + + setWorkflowResult(finalResult as typeof workflowResult); + setIsModalOpen(false); + } catch (error) { + const msg = error instanceof Error ? error.message : String(error); + console.error("Workflow failed:", msg); + setErrorMessage(msg); + } finally { + setIsLoading(false); + } + }; + + return ( +
+
Workflow Engine (v4)
+ + {output ? ( +
+

๐ŸŽ‰ Workflow Complete!

+ +
+
+

Step 1: Greeting

+

{output.greeting}

+
+ +
+

Step 2: Echo Transform

+

{output.echo}

+
+ +
+

Excitement Level

+
+
+ {output.excitementLevel}/10 +
+
+
+ +

+ Completed at {new Date(output.timestamp).toLocaleTimeString()} +

+ + +
+ ) : ( +
+

๐Ÿ”„ Workflow Engine Ready

+

This workflow demonstrates multi-step tool composition:

+
    +
  1. Greet a person
    2. +
  2. Add excitement level
    3. +
  3. Echo with uppercase transform
    4. +
  4. Combine all results
    5. +
+ +
+ )} + + {isModalOpen && ( +
setIsModalOpen(false)}> +
e.stopPropagation()}> +

Configure Workflow

+ +
+ + setWorkflowInput({ ...workflowInput, name: e.target.value })} + placeholder="Enter a name" + autoFocus + onKeyDown={(e) => e.key === "Enter" && handleRunWorkflow()} + /> +
+ +
+ + + setWorkflowInput({ + ...workflowInput, + excitement: parseInt(e.target.value), + }) + } + /> +
+ Calm (1) + Excited (10) +
+
+ + {errorMessage &&

{errorMessage}

} + +
+ + +
+
+
+ )} +
+ ); +} + +export default WorkflowWidget; diff --git a/examples/minimal/src/ui/styles.css b/examples/minimal/src/ui/styles.css index c0796b16..bdc2dac1 100644 --- a/examples/minimal/src/ui/styles.css +++ b/examples/minimal/src/ui/styles.css @@ -245,3 +245,251 @@ body { border-color: #555; color: #fff; } + +/* Workflow-specific styles */ +.workflow-result { + background: linear-gradient(135deg, #f093fb 0%, #f5576c 100%); +} + +.workflow-steps { + margin: 20px 0; + display: flex; + flex-direction: column; + gap: 16px; +} + +.step-result { + background: rgba(255, 255, 255, 0.15); + padding: 16px; + border-radius: 8px; + backdrop-filter: blur(10px); + text-align: left; +} + +.step-result h3 { + font-size: 0.875rem; + font-weight: 600; + margin-bottom: 8px; + text-transform: uppercase; + letter-spacing: 0.5px; + opacity: 0.9; +} + +.step-result p { + font-size: 1rem; + line-height: 1.4; +} + +.echo-text { + font-weight: 700; + letter-spacing: 1px; + font-size: 1.1rem; +} + +.excitement-meter { + position: relative; + height: 32px; + background: rgba(255, 255, 255, 0.2); + border-radius: 16px; + overflow: hidden; +} + +.excitement-fill { + position: absolute; + left: 0; + top: 0; + height: 100%; + background: linear-gradient(90deg, #4facfe 0%, #00f2fe 100%); + transition: width 0.3s ease; +} + +.excitement-label { + position: absolute; + left: 50%; + top: 50%; + transform: translate(-50%, -50%); + font-weight: 600; + font-size: 0.875rem; + z-index: 1; +} + +.workflow-description { + text-align: left; + margin: 16px 0; + padding-left: 24px; +} + +.workflow-description li { + margin: 8px 0; + line-height: 1.4; +} + +.greetings-list, +.features-list { + list-style: none; + padding: 0; + margin: 8px 0 0 0; +} + +.greetings-list li { + padding: 8px 12px; + margin: 4px 0; + background: rgba(255, 255, 255, 0.1); + border-radius: 6px; +} + +.features-list li { + padding: 4px 0; + font-size: 0.875rem; +} + +/* Advanced workflow modal */ +.modal-large { + min-width: 400px; + max-width: 500px; +} + +.form-group { + margin-bottom: 20px; +} + +.form-group label { + display: block; + margin-bottom: 8px; + font-weight: 600; + color: #333; + font-size: 0.875rem; +} + +.dark .form-group label { + color: #fff; +} + +.name-input-group { + display: flex; + gap: 8px; + margin-bottom: 8px; +} + +.name-input-group input { + flex: 1; + margin-bottom: 0; +} + +.remove-btn { + padding: 8px 12px; + background: #dc3545; + color: white; + border: none; + border-radius: 6px; + cursor: pointer; + font-size: 0.875rem; + transition: background 0.2s; +} + +.remove-btn:hover { + background: #c82333; +} + +.add-btn { + width: 100%; + padding: 8px; + background: #28a745; + color: white; + border: none; + border-radius: 6px; + cursor: pointer; + font-size: 0.875rem; + transition: background 0.2s; +} + +.add-btn:hover { + background: #218838; +} + +.radio-group { + display: flex; + gap: 16px; +} + +.radio-label { + display: flex; + align-items: center; + gap: 6px; + cursor: pointer; + font-size: 0.875rem; + color: #666; +} + +.dark .radio-label { + color: #aaa; +} + +.radio-label input[type="radio"] { + width: auto; + margin: 0; + cursor: pointer; +} + +.range-labels { + display: flex; + justify-content: space-between; + font-size: 0.75rem; + color: #666; + margin-top: 4px; +} + +.dark .range-labels { + color: #aaa; +} + +input[type="range"] { + width: 100%; + margin: 8px 0; + padding: 0; + -webkit-appearance: none; + appearance: none; + background: transparent; +} + +input[type="range"]::-webkit-slider-track { + width: 100%; + height: 6px; + background: #ddd; + border-radius: 3px; +} + +.dark input[type="range"]::-webkit-slider-track { + background: #444; +} + +input[type="range"]::-webkit-slider-thumb { + -webkit-appearance: none; + appearance: none; + width: 18px; + height: 18px; + background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); + border-radius: 50%; + cursor: pointer; + margin-top: -6px; +} + +input[type="range"]::-moz-range-track { + width: 100%; + height: 6px; + background: #ddd; + border-radius: 3px; +} + +.dark input[type="range"]::-moz-range-track { + background: #444; +} + +input[type="range"]::-moz-range-thumb { + width: 18px; + height: 18px; + background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); + border: none; + border-radius: 50%; + cursor: pointer; +} diff --git a/packages/core/README.md b/packages/core/README.md index 5ad19558..8bc6dd9a 100644 --- a/packages/core/README.md +++ b/packages/core/README.md @@ -14,6 +14,8 @@ MCP AppsKit Core is the server runtime for defining tools, validating inputs and - [Install](#install) - [Usage](#usage) - [Type-Safe Tool Definitions](#type-safe-tool-definitions) +- [API Versioning](#api-versioning) +- [Workflow Engine](#workflow-engine) - [Plugins, Middleware & Events](#plugins-middleware--events) - [Debug Logging](#debug-logging) - [OAuth 2.1 Authentication](#oauth-21-authentication) @@ -31,6 +33,7 @@ Interactive MCP apps often need to support multiple hosts with slightly differen - Single `createApp()` entry point for tools and UI definitions - **API Versioning**: Expose multiple API versions from a single app (e.g., `/v1/mcp`, `/v2/mcp`) +- **Workflow Engine**: Compose multi-step workflows with parallel execution, branching, and retry policies - Zod-powered validation with strong TypeScript inference - Unified metadata for MCP Apps and ChatGPT Apps - OAuth 2.1 bearer token validation with JWKS discovery @@ -470,6 +473,227 @@ app.getVersions(); // [] app.getVersion("v1"); // undefined ``` +## Workflow Engine + +The workflow engine enables composing multi-step workflows as MCP tools. Workflows support tool calls, custom logic, parallel execution, conditional branching, and configurable error handling. + +### Basic Workflow + +```ts +import { workflow, toolStep, customStep } from "@mcp-apps-kit/core"; +import { z } from "zod"; + +const orderWorkflow = workflow("process_order") + .describe("Process a customer order end-to-end") + .input({ + orderId: z.string(), + customerId: z.string(), + }) + .output({ + success: z.boolean(), + receiptId: z.string().optional(), + }) + .step("validate", toolStep("validate_order")) + .step("payment", toolStep("process_payment")) + .step( + "fulfill", + customStep(async (ctx) => { + // Access previous step outputs + const paymentResult = ctx.outputs.payment; + return { success: true, receiptId: paymentResult.receiptId }; + }) + ) + .build(); + +// Register as a tool +const app = createApp({ + name: "order-service", + tools: { process_order: orderWorkflow }, +}); +``` + +### Step Types + +#### Tool Steps + +Call other registered tools in the same app: + +```ts +.step("validate", toolStep("validate_order")) + +// With input mapping +.step("payment", toolStep("process_payment", { + mapInput: (ctx) => ({ + amount: ctx.outputs.validate.total, + customerId: ctx.input.customerId, + }), +})) +``` + +#### Custom Steps + +Execute arbitrary async logic: + +```ts +.step("enrich", customStep(async (ctx) => { + const userData = await fetchUserData(ctx.input.userId); + return { ...ctx.input, userData }; +})) +``` + +#### External Steps + +Call tools on external MCP servers: + +```ts +import { externalStep } from "@mcp-apps-kit/core"; + +.step("weather", externalStep({ + server: "mcp://weather-service", + tool: "get_forecast", + mapInput: (ctx) => ({ + location: ctx.input.destination, + date: ctx.input.date, + }), +})) +``` + +### Parallel Execution + +Run multiple steps concurrently: + +```ts +.parallel("notifications", [ + toolStep("send_email"), + toolStep("send_sms"), + toolStep("log_event"), +]) +``` + +### Conditional Branching + +Execute different paths based on runtime conditions: + +```ts +.branch("shipping_method", { + when: (ctx) => ctx.outputs.validate.isDigital, + then: [customStep(async () => ({ delivered: true }))], + else: [toolStep("create_shipment"), toolStep("notify_warehouse")], +}) +``` + +### Step Configuration + +Configure retry, timeout, and error handling per step: + +```ts +.step("payment", toolStep("process_payment"), { + // Retry configuration + retry: { + maxAttempts: 3, + delay: 1000, // Initial delay in ms + backoff: "exponential", // "linear" | "exponential" + maxDelay: 10000, // Cap for exponential backoff + }, + + // Timeout + timeout: 30000, // 30 seconds + + // Error handling + onError: "skip", // "fail" | "skip" | custom handler +}) + +// Custom error handler +.step("optional", toolStep("optional_step"), { + onError: async (error, ctx) => { + console.log(`Step failed: ${error.message}`); + return { fallbackResult: true }; // Return fallback value + }, +}) +``` + +### Attaching UI to Workflows + +Workflows can have UI widgets just like regular tools: + +```ts +const workflowUI = defineUI({ + name: "Order Status", + html: "./dist/order-widget.html", +}); + +const orderWorkflow = workflow("process_order") + .describe("Process order") + .input({ orderId: z.string() }) + .ui(workflowUI) + .step("validate", toolStep("validate_order")) + .build(); +``` + +### Production Lifecycle Management + +Workflows automatically detect your environment and use the appropriate executor manager: + +**Traditional Servers** (Node.js, Express): + +- Global singleton with persistent pooling +- Background cleanup of idle executors (10 min TTL) +- LRU eviction when pool reaches 100 executors +- Graceful shutdown via `server.stop()` + +**Edge/Serverless** (Supabase, Vercel, Cloudflare, AWS Lambda): + +- Fresh manager per invocation (no singleton) +- Smaller pool size (10 executors, memory-constrained) +- No background timers +- Auto-cleanup on function exit + +```ts +// Traditional server - automatic cleanup on stop +const server = createApp({ tools: { myWorkflow } }); +await server.start(); +await server.stop(); // Cleans up all workflow resources + +// Advanced configuration +import { ExecutorManager } from "@mcp-apps-kit/core"; + +const manager = ExecutorManager.getInstance({ + maxExecutors: 200, // Pool size for high-traffic apps + executorTTL: 5 * 60 * 1000, // 5 minute idle timeout + autoCleanup: true, // Enable automatic cleanup + cleanupInterval: 60 * 1000, // Cleanup every minute +}); + +// Get statistics +const stats = manager.getStats(); +console.log(`Active: ${stats.activeExecutors}, Total: ${stats.totalExecutors}`); +``` + +### Workflow Errors + +The workflow engine provides specific error types for debugging: + +```ts +import { + WorkflowError, // Base class for all workflow errors + WorkflowExecutionError, // Step execution failures + StepTimeoutError, // Step timeout exceeded + ExternalToolError, // External MCP call failures + WorkflowValidationError, // Input/output validation failures + WorkflowDefinitionError, // Invalid workflow configuration +} from "@mcp-apps-kit/core"; + +try { + await orderWorkflow.handler(input, context); +} catch (error) { + if (error instanceof StepTimeoutError) { + console.log(`Step timed out: ${error.stepName}`); + } else if (error instanceof ExternalToolError) { + console.log(`External call failed: ${error.serverUri}`); + } +} +``` + ## Plugins, Middleware & Events ### Plugins @@ -997,6 +1221,8 @@ const app = createApp({ Key exports include: - `createApp`, `tool`, `defineTool`, `defineUI` +- `workflow`, `toolStep`, `customStep`, `externalStep` +- `ExecutorManager`, `ExternalToolClient` - `createPlugin`, `loggingPlugin` - `debugLogger`, `ClientToolsFromCore` - `Middleware`, `TypedEventEmitter` diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index d5365d6f..82239c9d 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -223,3 +223,55 @@ export type { ToolBuilderComplete, ToolBuilderConfigurable, } from "./builder/tool-builder"; + +// ============================================================================= +// WORKFLOW ENGINE +// ============================================================================= + +export { + workflow, + toolStep, + customStep, + externalStep, + WorkflowExecutor, + ExternalToolClient, + ExecutorManager, + EdgeExecutorManager, + WorkflowError, + WorkflowExecutionError, + WorkflowDefinitionError, + ToolResponseValidationError, + StepTimeoutError, + ExternalToolError, + WorkflowValidationError, +} from "./workflow"; + +export type { + WorkflowContext, + ToolCaller, + ExternalToolCaller, + ToolValidator, + RetryConfig, + ErrorHandler, + ErrorHandling, + StepConfig, + ToolStep, + CustomStep, + ExternalStep, + ParallelStep, + BranchStep, + Step, + NamedStep, + WorkflowDefinition, + StepExecutionResult, + WorkflowExecutionResult, + WorkflowBuilderInitial, + WorkflowBuilderWithDescription, + WorkflowBuilderWithInput, + WorkflowBuilderWithOutput, + WorkflowBuilderWithSteps, + ExternalStepConfig, + ExternalToolClientConfig, + ExecutorManagerConfig, + EdgeExecutorManagerConfig, +} from "./workflow"; diff --git a/packages/core/src/server/index.ts b/packages/core/src/server/index.ts index 14db093b..9145c152 100644 --- a/packages/core/src/server/index.ts +++ b/packages/core/src/server/index.ts @@ -393,7 +393,18 @@ export function createServerInstance( }, stop: async () => { - return new Promise((resolve) => { + // Shutdown workflow executors first, but always close HTTP server + let executorError: Error | undefined; + try { + const { ExecutorManager } = await import("../workflow/executor-manager.js"); + await ExecutorManager.getInstance().shutdown(); + } catch (error) { + // Capture error but continue to close HTTP server + executorError = error instanceof Error ? error : new Error(String(error)); + } + + // Always stop the HTTP server, even if executor shutdown failed + await new Promise((resolve) => { if (httpServer) { httpServer.close(() => { httpServer = undefined; @@ -404,6 +415,11 @@ export function createServerInstance( resolve(); } }); + + // Re-throw executor error after server is closed + if (executorError) { + throw executorError; + } }, handler: (): ExpressMiddleware => { @@ -745,8 +761,30 @@ function registerTools( // Create state map for middleware const state = new Map(); - // Create full context with state - const context: ToolContext = { ...baseContext, state }; + // Create internal tool caller for workflows + const internalToolCaller = async ( + targetToolName: string, + targetInput: unknown, + targetContext: ToolContext + ): Promise => { + const targetToolDef = tools[targetToolName]; + if (!targetToolDef) { + throw new Error(`Tool "${targetToolName}" not found`); + } + + // Validate input + const validatedInput = targetToolDef.input.parse(targetInput); + + // Execute the target tool's handler with the provided context + return await targetToolDef.handler(validatedInput, targetContext); + }; + + // Create full context with state and internal tool caller + const context: ToolContext = { + ...baseContext, + state, + _internalToolCaller: internalToolCaller, + }; contextForErrorHandling = context; // Emit tool:called event diff --git a/packages/core/src/types/tools.ts b/packages/core/src/types/tools.ts index b4aa2882..cddaa731 100644 --- a/packages/core/src/types/tools.ts +++ b/packages/core/src/types/tools.ts @@ -169,6 +169,18 @@ export interface ToolContext { * ``` */ state?: Map; + + /** + * Internal tool caller for workflows. + * Allows workflows to call other tools in the same app. + * + * @internal + */ + _internalToolCaller?: ( + toolName: string, + input: unknown, + context: ToolContext + ) => Promise; } /** diff --git a/packages/core/src/workflow/errors.ts b/packages/core/src/workflow/errors.ts new file mode 100644 index 00000000..aac4701b --- /dev/null +++ b/packages/core/src/workflow/errors.ts @@ -0,0 +1,98 @@ +/** + * Workflow engine error types + * + * @module workflow/errors + */ + +/** + * Base error class for workflow-related errors + */ +export class WorkflowError extends Error { + constructor( + message: string, + public readonly code: string, + public readonly details?: Record + ) { + super(message); + this.name = "WorkflowError"; + } +} + +/** + * Error thrown when workflow execution fails + */ +export class WorkflowExecutionError extends WorkflowError { + constructor( + message: string, + public readonly stepName?: string, + details?: Record + ) { + super(message, "WORKFLOW_EXECUTION_ERROR", details); + this.name = "WorkflowExecutionError"; + } +} + +/** + * Error thrown when a step times out + */ +export class StepTimeoutError extends WorkflowError { + constructor( + public readonly stepName: string, + public readonly timeout: number + ) { + super(`Step "${stepName}" timed out after ${timeout}ms`, "STEP_TIMEOUT", { + stepName, + timeout, + }); + this.name = "StepTimeoutError"; + } +} + +/** + * Error thrown when external tool call fails + */ +export class ExternalToolError extends WorkflowError { + constructor( + message: string, + public readonly server: string, + public readonly toolName: string, + details?: Record + ) { + super(message, "EXTERNAL_TOOL_ERROR", { ...details, server, toolName }); + this.name = "ExternalToolError"; + } +} + +/** + * Error thrown when workflow validation fails during build + */ +export class WorkflowValidationError extends WorkflowError { + constructor(message: string, details?: Record) { + super(message, "WORKFLOW_VALIDATION_ERROR", details); + this.name = "WorkflowValidationError"; + } +} + +/** + * Error thrown when workflow definition is invalid at runtime + */ +export class WorkflowDefinitionError extends WorkflowError { + constructor(message: string, details?: Record) { + super(message, "WORKFLOW_DEFINITION_ERROR", details); + this.name = "WorkflowDefinitionError"; + } +} + +/** + * Error thrown when tool response validation fails + */ +export class ToolResponseValidationError extends WorkflowError { + constructor( + message: string, + public readonly toolName: string, + details?: Record + ) { + super(message, "TOOL_RESPONSE_VALIDATION_ERROR", { ...details, toolName }); + this.name = "ToolResponseValidationError"; + } +} diff --git a/packages/core/src/workflow/executor-manager-edge.ts b/packages/core/src/workflow/executor-manager-edge.ts new file mode 100644 index 00000000..79e3b7b4 --- /dev/null +++ b/packages/core/src/workflow/executor-manager-edge.ts @@ -0,0 +1,337 @@ +/** + * Edge-optimized Executor Manager for stateless serverless environments + * + * Designed for Supabase Edge Functions, Cloudflare Workers, Vercel Edge, etc. + * where each invocation is isolated and short-lived. + * + * @module workflow/executor-manager-edge + */ + +import { WorkflowExecutor } from "./executor"; +import type { WorkflowDefinition } from "./types"; + +// Type declarations for edge runtime globals +declare const Deno: + | { + env?: { + get?: (key: string) => string | undefined; + }; + } + | undefined; + +// Worker global scope type for Deno/Workers +interface WorkerGlobalScope { + addEventListener(type: string, listener: () => void): void; +} +declare const self: WorkerGlobalScope | undefined; + +// Global registry for cleanup handlers +const globalCleanupRegistry: Set = + (globalThis as unknown as { __edgeExecutorManagers?: Set }) + .__edgeExecutorManagers ?? new Set(); +( + globalThis as unknown as { __edgeExecutorManagers: Set } +).__edgeExecutorManagers = globalCleanupRegistry; + +let cleanupHandlersRegistered = false; + +// ============================================================================= +// CONFIGURATION +// ============================================================================= + +/** + * Configuration for EdgeExecutorManager + */ +export interface EdgeExecutorManagerConfig { + /** + * Maximum number of executors to cache per function invocation + * @default 10 (smaller than standard manager due to memory constraints) + */ + maxExecutors?: number; +} + +// ============================================================================= +// EDGE EXECUTOR MANAGER +// ============================================================================= + +/** + * Edge-optimized executor manager for serverless/edge environments + * + * Key differences from standard ExecutorManager: + * - No singleton (each invocation gets fresh instance) + * - No background cleanup timers (rely on function termination) + * - Simplified pooling (function lifetime is short) + * - Automatic cleanup on function exit via process handlers + * - Smaller default pool size (edge functions have memory limits) + * + * @example Supabase Edge Function + * ```typescript + * import { serve } from "https://deno.land/std/http/server.ts"; + * import { createApp, EdgeExecutorManager } from "@mcp-apps-kit/core"; + * + * const app = createApp({ + * name: "my-edge-app", + * tools: { myWorkflow }, + * }); + * + * // Configure for edge environment + * EdgeExecutorManager.configureDefaults({ + * maxExecutors: 5, // Smaller pool for memory-constrained edge + * }); + * + * serve(async (req) => { + * const response = await app.handleRequest(req); + * return response; + * }); + * ``` + * + * @example Manual cleanup for long-lived edge workers (Cloudflare Durable Objects) + * ```typescript + * export class WorkflowWorker { + * private manager = new EdgeExecutorManager({ maxExecutors: 5 }); + * + * async fetch(request: Request) { + * try { + * // Handle request... + * } finally { + * // Cleanup if this worker instance is shutting down + * if (request.headers.get("X-Worker-Shutdown")) { + * await this.manager.shutdown(); + * } + * } + * } + * } + * ``` + */ +export class EdgeExecutorManager { + private static defaultConfig: Required = { + maxExecutors: 10, // Smaller for memory-constrained edge + }; + + private executors: Map = new Map(); + private lastUsed: Map = new Map(); + private config: Required; + + /** + * Configure default settings for all EdgeExecutorManager instances + * + * Call this once at application startup in your edge function entry point. + */ + static configureDefaults(config: Partial): void { + // Validate maxExecutors if provided + const validatedConfig = { ...config }; + if (validatedConfig.maxExecutors !== undefined) { + validatedConfig.maxExecutors = Math.max(1, Math.floor(validatedConfig.maxExecutors)); + } + + EdgeExecutorManager.defaultConfig = { + ...EdgeExecutorManager.defaultConfig, + ...validatedConfig, + }; + } + + constructor(config?: EdgeExecutorManagerConfig) { + // Validate and merge config + const validatedConfig = config ? { ...config } : undefined; + if (validatedConfig?.maxExecutors !== undefined) { + validatedConfig.maxExecutors = Math.max(1, Math.floor(validatedConfig.maxExecutors)); + } + + this.config = { + ...EdgeExecutorManager.defaultConfig, + ...validatedConfig, + }; + + // Add this instance to global registry + globalCleanupRegistry.add(this); + + // Register cleanup on process exit (works in most edge runtimes) + // This ensures connections are closed when the function terminates + this.registerCleanupHooks(); + } + + /** + * Get or create an executor for a workflow + * + * In edge environments, this creates a new executor per invocation + * unless the workflow was already used in the same function invocation. + */ + getOrCreate(definition: WorkflowDefinition): WorkflowExecutor { + const key = definition.name; + const existing = this.executors.get(key); + + if (existing) { + // Update last used timestamp for LRU tracking + this.lastUsed.set(key, Date.now()); + return existing; + } + + // LRU eviction: if full, remove least recently used + if (this.executors.size >= this.config.maxExecutors) { + let oldestKey: string | undefined; + let oldestTime = Infinity; + + // Find the executor with the oldest lastUsed timestamp + for (const [executorKey, timestamp] of this.lastUsed.entries()) { + if (timestamp < oldestTime) { + oldestTime = timestamp; + oldestKey = executorKey; + } + } + + if (oldestKey) { + const executor = this.executors.get(oldestKey); + if (executor) { + // Fire and forget cleanup + executor.close().catch(() => { + // Ignore errors during eviction + }); + } + this.executors.delete(oldestKey); + this.lastUsed.delete(oldestKey); + } + } + + const executor = new WorkflowExecutor(definition); + this.executors.set(key, executor); + this.lastUsed.set(key, Date.now()); + + return executor; + } + + /** + * Mark an executor as in-use (no-op in edge environments) + * + * Edge functions are short-lived, so reference counting isn't needed. + * This method exists for API compatibility with ExecutorManager. + */ + markInUse(_workflowName: string): void { + // No-op: edge functions terminate quickly, no need for reference counting + } + + /** + * Mark an executor as idle (no-op in edge environments) + * + * Edge functions are short-lived, so reference counting isn't needed. + * This method exists for API compatibility with ExecutorManager. + */ + markIdle(_workflowName: string): void { + // No-op: edge functions terminate quickly, no need for reference counting + } + + /** + * Get statistics about the executor pool + */ + getStats(): { totalExecutors: number } { + return { + totalExecutors: this.executors.size, + }; + } + + /** + * Synchronous cleanup for use in synchronous exit handlers + * + * Performs best-effort cleanup without waiting for async operations. + * Used by process.on("exit") which cannot wait for promises. + */ + private shutdownSync(): void { + // Clear maps immediately + this.executors.clear(); + this.lastUsed.clear(); + + // Remove from global registry + globalCleanupRegistry.delete(this); + } + + /** + * Manually cleanup all executors + * + * In most edge environments, this is called automatically via + * process exit handlers. You typically don't need to call this manually. + */ + async shutdown(): Promise { + const closePromises: Promise[] = []; + + for (const executor of this.executors.values()) { + closePromises.push( + executor.close().catch(() => { + // Ignore errors during shutdown + }) + ); + } + + await Promise.all(closePromises); + this.executors.clear(); + this.lastUsed.clear(); + + // Remove this instance from global registry + globalCleanupRegistry.delete(this); + } + + /** + * Register cleanup hooks for the edge runtime + * + * Different runtimes have different exit signals: + * - Node.js: process.on('exit') is synchronous, SIGTERM/SIGINT are async + * - Deno: self.addEventListener('unload') is synchronous + * - Cloudflare Workers: scheduled cleanup in fetch handler + * + * Uses a global guard to ensure handlers are registered exactly once, + * and each handler invokes shutdown() on all registered instances. + */ + private registerCleanupHooks(): void { + // Only register handlers once globally + if (cleanupHandlersRegistered) return; + cleanupHandlersRegistered = true; + + // Try to detect runtime and register appropriate hooks + const isNode = typeof process !== "undefined" && process.versions?.node; + const isDeno = typeof Deno !== "undefined"; + + // Synchronous cleanup for exit handler (can't wait for promises) + const cleanupAllSync = () => { + for (const manager of globalCleanupRegistry) { + manager.shutdownSync(); + } + }; + + // Async cleanup for signal handlers (can await promises then exit) + const cleanupAllAsync = async (signal: string) => { + try { + const shutdowns = Array.from(globalCleanupRegistry).map((manager) => + manager.shutdown().catch(() => { + // Ignore errors during shutdown + }) + ); + await Promise.all(shutdowns); + } finally { + // Exit after cleanup completes + if (signal === "SIGTERM" || signal === "SIGINT") { + process.exit(0); + } + } + }; + + if (isNode) { + // Node.js edge functions (Vercel, Netlify) + // process.on("exit") is synchronous - use sync cleanup + process.on("exit", cleanupAllSync); + + // SIGTERM/SIGINT allow async cleanup before exit + process.on("SIGTERM", () => { + void cleanupAllAsync("SIGTERM"); + }); + process.on("SIGINT", () => { + void cleanupAllAsync("SIGINT"); + }); + } else if (isDeno) { + // Deno edge functions (Supabase, Deno Deploy) + // unload event is synchronous - use sync cleanup + if (typeof self !== "undefined" && "addEventListener" in self) { + self.addEventListener("unload", cleanupAllSync); + } + } + // Cloudflare Workers don't have exit hooks - cleanup must be manual + // or rely on the function instance being terminated + } +} diff --git a/packages/core/src/workflow/executor-manager.ts b/packages/core/src/workflow/executor-manager.ts new file mode 100644 index 00000000..2de30999 --- /dev/null +++ b/packages/core/src/workflow/executor-manager.ts @@ -0,0 +1,379 @@ +/** + * Executor Manager - Production-ready lifecycle management for workflow executors + * + * @module workflow/executor-manager + */ + +import { WorkflowExecutor } from "./executor"; +import type { WorkflowDefinition } from "./types"; +import { debugLogger } from "../debug/logger"; + +/** + * Configuration for the ExecutorManager + */ +export interface ExecutorManagerConfig { + /** + * Maximum number of executors to cache + * @default 100 + */ + maxExecutors?: number; + + /** + * Time-to-live for idle executors in milliseconds + * @default 600000 (10 minutes) + */ + executorTTL?: number; + + /** + * Enable automatic cleanup of idle executors + * @default true + */ + autoCleanup?: boolean; + + /** + * Cleanup interval in milliseconds + * @default 60000 (1 minute) + */ + cleanupInterval?: number; +} + +interface ManagedExecutor { + executor: WorkflowExecutor; + definition: WorkflowDefinition; + lastUsed: number; + activeInvocations: number; +} + +/** + * ExecutorManager provides centralized lifecycle management for workflow executors + * + * Features: + * - Executor pooling and reuse across invocations + * - Automatic cleanup of idle executors + * - LRU eviction when pool is full + * - Reference counting to prevent premature cleanup + * - Graceful shutdown with resource cleanup + * + * @example + * ```typescript + * // Get the global instance + * const manager = ExecutorManager.getInstance(); + * + * // Get or create an executor + * const executor = manager.getOrCreate(workflowDef); + * + * // Mark as in-use during execution + * manager.markInUse(workflowDef.name); + * try { + * const result = await executor.execute(input, context); + * } finally { + * manager.markIdle(workflowDef.name); + * } + * + * // Cleanup on server shutdown + * await manager.shutdown(); + * ``` + */ +export class ExecutorManager { + private static instance: ExecutorManager | undefined; + private executors: Map = new Map(); + private cleanupTimer: ReturnType | undefined; + private config: Required; + private isShuttingDown = false; + + /** + * Get the global ExecutorManager instance + * + * This ensures a single manager is shared across the entire application, + * maximizing executor reuse and connection pooling efficiency. + */ + static getInstance(config?: ExecutorManagerConfig): ExecutorManager { + ExecutorManager.instance ??= new ExecutorManager(config); + return ExecutorManager.instance; + } + + /** + * Reset the global instance (primarily for testing) + * @internal + */ + static resetInstance(): void { + if (ExecutorManager.instance) { + // Don't await - fire and forget cleanup + ExecutorManager.instance.shutdown().catch(() => { + // Ignore errors during reset + }); + ExecutorManager.instance = undefined; + } + } + + private constructor(config: ExecutorManagerConfig = {}) { + this.config = { + maxExecutors: config.maxExecutors ?? 100, + executorTTL: config.executorTTL ?? 10 * 60 * 1000, // 10 minutes + autoCleanup: config.autoCleanup ?? true, + cleanupInterval: config.cleanupInterval ?? 60 * 1000, // 1 minute + }; + + if (this.config.autoCleanup) { + this.startCleanupTimer(); + } + } + + /** + * Get or create an executor for a workflow definition + * + * If an executor already exists for this workflow name, it is reused. + * Otherwise, a new executor is created and cached. + * + * @param definition - Workflow definition + * @returns Workflow executor instance + */ + getOrCreate(definition: WorkflowDefinition): WorkflowExecutor { + if (this.isShuttingDown) { + throw new Error("ExecutorManager is shutting down, cannot create new executors"); + } + + const key = definition.name; + const existing = this.executors.get(key); + + if (existing) { + // Update last used timestamp + existing.lastUsed = Date.now(); + return existing.executor; + } + + // Evict old executors if cache is full + if (this.executors.size >= this.config.maxExecutors) { + const evicted = this.evictIdleLRU(); + if (!evicted) { + // All executors are active - fail fast rather than break running workflows + throw new Error( + `ExecutorManager capacity exceeded: all ${this.config.maxExecutors} executors are active. ` + + `Increase maxExecutors or wait for active workflows to complete.` + ); + } + } + + // Create new executor + const executor = new WorkflowExecutor(definition); + this.executors.set(key, { + executor, + definition, + lastUsed: Date.now(), + activeInvocations: 0, + }); + + debugLogger.debug(`Created workflow executor: ${key}`, { + totalExecutors: this.executors.size, + }); + + return executor; + } + + /** + * Mark an executor as in-use (increment reference count) + * + * Call this before executing a workflow to prevent the executor + * from being cleaned up during execution. + * + * @param workflowName - Name of the workflow + */ + markInUse(workflowName: string): void { + const managed = this.executors.get(workflowName); + if (managed) { + managed.activeInvocations++; + } + } + + /** + * Mark an executor as idle (decrement reference count) + * + * Call this after workflow execution completes to allow + * the executor to be cleaned up if idle for too long. + * + * @param workflowName - Name of the workflow + */ + markIdle(workflowName: string): void { + const managed = this.executors.get(workflowName); + if (managed) { + managed.activeInvocations = Math.max(0, managed.activeInvocations - 1); + managed.lastUsed = Date.now(); + } + } + + /** + * Get statistics about the executor pool + */ + getStats(): { + totalExecutors: number; + activeExecutors: number; + idleExecutors: number; + oldestExecutorAge: number; + } { + const now = Date.now(); + let activeCount = 0; + let oldestAge = 0; + + for (const managed of this.executors.values()) { + if (managed.activeInvocations > 0) { + activeCount++; + } + const age = now - managed.lastUsed; + if (age > oldestAge) { + oldestAge = age; + } + } + + return { + totalExecutors: this.executors.size, + activeExecutors: activeCount, + idleExecutors: this.executors.size - activeCount, + oldestExecutorAge: oldestAge, + }; + } + + /** + * Manually trigger cleanup of idle executors + * + * This is called automatically by the cleanup timer if autoCleanup is enabled. + * You can also call it manually for fine-grained control. + */ + async cleanup(): Promise { + const now = Date.now(); + const toCleanup: string[] = []; + + // Find idle executors past TTL + for (const [key, managed] of this.executors.entries()) { + const isIdle = managed.activeInvocations === 0; + const isPastTTL = now - managed.lastUsed > this.config.executorTTL; + + if (isIdle && isPastTTL) { + toCleanup.push(key); + } + } + + // Cleanup executors + for (const key of toCleanup) { + await this.closeExecutor(key); + } + + if (toCleanup.length > 0) { + debugLogger.debug(`Cleaned up ${toCleanup.length} idle executor(s)`, { + remaining: this.executors.size, + }); + } + } + + /** + * Shutdown the executor manager and cleanup all resources + * + * This should be called during server shutdown to ensure all + * external MCP connections are properly closed. + * + * @param force - If true, close executors even if they have active invocations + */ + async shutdown(force = false): Promise { + this.isShuttingDown = true; + + // Stop the cleanup timer + if (this.cleanupTimer) { + clearInterval(this.cleanupTimer); + this.cleanupTimer = undefined; + } + + debugLogger.info("Shutting down ExecutorManager", { + totalExecutors: this.executors.size, + force, + }); + + // Close all executors + const closePromises: Promise[] = []; + for (const [key, managed] of this.executors.entries()) { + if (force || managed.activeInvocations === 0) { + closePromises.push( + this.closeExecutor(key).catch((error: unknown) => { + debugLogger.error(`Failed to close executor: ${key}`, { error }); + }) + ); + } else { + debugLogger.warn(`Executor has active invocations, skipping: ${key}`, { + activeInvocations: managed.activeInvocations, + }); + } + } + + await Promise.all(closePromises); + + debugLogger.info("ExecutorManager shutdown complete", { + remainingExecutors: this.executors.size, + }); + } + + /** + * Evict the least recently used idle executor + * + * Only evicts idle executors (activeInvocations === 0) to avoid + * breaking running workflows. Returns false if no idle executor + * is available for eviction. + * + * @returns true if an executor was evicted, false if all are active + */ + private evictIdleLRU(): boolean { + let oldestKey: string | undefined; + let oldestTime = Infinity; + + // Find the oldest idle executor only + for (const [key, managed] of this.executors.entries()) { + if (managed.activeInvocations === 0 && managed.lastUsed < oldestTime) { + oldestTime = managed.lastUsed; + oldestKey = key; + } + } + + // If no idle executor found, don't evict active ones + if (!oldestKey) { + debugLogger.debug("No idle executors available for eviction"); + return false; + } + + debugLogger.debug(`Evicting LRU idle executor: ${oldestKey}`); + // Fire and forget - don't block on cleanup + this.closeExecutor(oldestKey).catch(() => { + // Ignore errors during eviction + }); + return true; + } + + /** + * Close and remove an executor + */ + private async closeExecutor(key: string): Promise { + const managed = this.executors.get(key); + if (!managed) return; + + this.executors.delete(key); + + try { + await managed.executor.close(); + } catch (error) { + debugLogger.error(`Error closing executor: ${key}`, { error }); + throw error; + } + } + + /** + * Start the automatic cleanup timer + */ + private startCleanupTimer(): void { + this.cleanupTimer = setInterval(() => { + this.cleanup().catch((error: unknown) => { + debugLogger.error("Error during automatic executor cleanup", { error }); + }); + }, this.config.cleanupInterval); + + // Don't prevent Node.js from exiting + if (this.cleanupTimer.unref) { + this.cleanupTimer.unref(); + } + } +} diff --git a/packages/core/src/workflow/executor.ts b/packages/core/src/workflow/executor.ts new file mode 100644 index 00000000..e9d4500f --- /dev/null +++ b/packages/core/src/workflow/executor.ts @@ -0,0 +1,527 @@ +/** + * Workflow executor - runtime execution engine + * + * @module workflow/executor + */ + +import type { z } from "zod"; +import type { ToolContext } from "../types/tools"; +import type { + WorkflowDefinition, + WorkflowContext, + WorkflowExecutionResult, + StepExecutionResult, + Step, + StepConfig, + ToolStep, + CustomStep, + ExternalStep, + ParallelStep, + BranchStep, + NamedStep, + RetryConfig, + ErrorHandling, +} from "./types"; +import { + WorkflowExecutionError, + StepTimeoutError, + ExternalToolError, + WorkflowDefinitionError, + ToolResponseValidationError, +} from "./errors"; +import { ExternalToolClient } from "./external-client"; + +/** + * Type for a validator function or Zod schema + */ +type Validator = z.ZodType | ((value: unknown) => T); + +/** + * Type guard to check if a validator is a Zod schema + */ +function isZodSchema(validator: Validator): validator is z.ZodType { + return ( + typeof validator === "object" && + validator !== null && + "parse" in validator && + typeof validator.parse === "function" + ); +} + +// ============================================================================= +// WORKFLOW EXECUTOR +// ============================================================================= + +/** + * Workflow execution engine + * + * Handles runtime execution of workflow steps with: + * - Accumulated context management + * - Retry logic with configurable backoff + * - Error handling per step configuration + * - Parallel execution + * - Conditional branching + */ +export class WorkflowExecutor { + private definition: WorkflowDefinition; + private externalClient: ExternalToolClient; + + constructor(definition: WorkflowDefinition) { + this.definition = definition; + this.externalClient = new ExternalToolClient(); + } + + /** + * Close the executor and release resources + * + * This closes all external MCP client connections. + * Call this method when the executor is no longer needed. + */ + async close(): Promise { + await this.externalClient.closeAll(); + } + + /** + * Execute the workflow with the given input + * + * @param input - Workflow input matching the input schema + * @param toolContext - MCP tool context + * @returns Workflow execution result + */ + async execute(input: unknown, toolContext: ToolContext): Promise { + const startTime = Date.now(); + const stepResults: StepExecutionResult[] = []; + const outputs: Record = {}; + + // Validate input + const validatedInput = this.definition.inputSchema.parse(input); + + // Create workflow context + const context: WorkflowContext = { + input: validatedInput, + outputs, + toolContext, + callTool: async ( + toolName: string, + toolInput: unknown, + validator?: Validator + ): Promise => { + // Use the internal tool caller from context if available + if (!toolContext._internalToolCaller) { + throw new WorkflowExecutionError( + `Internal tool caller not available. Cannot call tool: ${toolName}`, + undefined, + { toolName, input: toolInput } + ); + } + + // Call the tool via the internal caller + const result = await toolContext._internalToolCaller(toolName, toolInput, toolContext); + + // Validate the result if a validator is provided + if (validator) { + return this.validateToolResponse(result, toolName, validator); + } + + // No validator provided - return as unknown (caller is responsible for validation) + return result as TOutput; + }, + callExternalTool: async ( + server: string, + toolName: string, + toolInput: unknown, + validator?: Validator + ): Promise => { + const result = await this.externalClient.callTool(server, toolName, toolInput); + + // Validate the result if a validator is provided + if (validator) { + return this.validateToolResponse(result, `${server}:${toolName}`, validator); + } + + // No validator provided - return as unknown (caller is responsible for validation) + return result as TOutput; + }, + }; + + // Execute steps sequentially + for (const namedStep of this.definition.steps) { + const result = await this.executeNamedStep(namedStep, context); + stepResults.push(result); + + // Add step output to context (even if skipped, add undefined) + outputs[namedStep.name] = result.output; + } + + // Validate output if schema is provided + let finalOutput: unknown = outputs; + if (this.definition.outputSchema) { + // Check if workflow has no steps but declares an output schema + if (this.definition.steps.length === 0) { + throw new WorkflowDefinitionError( + `Workflow declares an outputSchema but has no steps to produce that output. ` + + `Add at least one step to the workflow or remove the outputSchema.`, + { + hasOutputSchema: true, + stepsCount: this.definition.steps.length, + } + ); + } + + // If output schema is defined, use the last step's output as the final output + // This allows the workflow to define its final output structure in the last step + const lastStep = this.definition.steps[this.definition.steps.length - 1]; + const lastStepOutput = lastStep ? outputs[lastStep.name] : outputs; + finalOutput = this.definition.outputSchema.parse(lastStepOutput); + } + + return { + output: finalOutput, + stepResults, + duration: Date.now() - startTime, + success: true, + }; + } + + /** + * Validate a tool response using the provided validator + */ + private validateToolResponse(result: unknown, toolName: string, validator: Validator): T { + try { + // Use type guard to check if validator is a Zod schema + if (isZodSchema(validator)) { + return validator.parse(result); + } + // Otherwise, it must be a validation function + if (typeof validator === "function") { + return validator(result); + } + throw new Error("Invalid validator: must be a Zod schema or validation function"); + } catch (error) { + throw new ToolResponseValidationError( + `Tool response validation failed for "${toolName}": ${(error as Error).message}`, + toolName, + { originalError: error, response: result } + ); + } + } + + /** + * Execute a named step + */ + private async executeNamedStep( + namedStep: NamedStep, + context: WorkflowContext + ): Promise { + const startTime = Date.now(); + let retries = 0; + + try { + // Execute step with retry logic, passing step name for error messages + const output = await this.executeStepWithRetry( + namedStep.name, + namedStep.step, + context, + (attempt) => { + retries = attempt; + } + ); + + return { + name: namedStep.name, + output, + duration: Date.now() - startTime, + retries, + }; + } catch (error) { + // Step failed after retries + const stepError = error as Error; + + // Get error handling strategy + const errorHandling = this.getErrorHandling(namedStep.step); + + if (errorHandling === "skip") { + // Skip this step and continue + return { + name: namedStep.name, + output: undefined, + duration: Date.now() - startTime, + retries, + error: stepError, + skipped: true, + }; + } else if (typeof errorHandling === "function") { + // Custom error handler + try { + const recoveryValue = await errorHandling(stepError, context); + return { + name: namedStep.name, + output: recoveryValue, + duration: Date.now() - startTime, + retries, + error: stepError, + }; + } catch (handlerError) { + // Error handler failed, re-throw original error + throw new WorkflowExecutionError( + `Step "${namedStep.name}" failed and error handler also failed: ${stepError.message}`, + namedStep.name, + { originalError: stepError, handlerError } + ); + } + } else { + // "fail" - propagate error + throw new WorkflowExecutionError( + `Step "${namedStep.name}" failed: ${stepError.message}`, + namedStep.name, + { originalError: stepError } + ); + } + } + } + + /** + * Execute a step with retry logic + * + * @param onRetry - Callback invoked with the retry number (1, 2, 3, ...) when a retry is about to be performed + */ + private async executeStepWithRetry( + stepName: string, + step: Step, + context: WorkflowContext, + onRetry: (retryNumber: number) => void + ): Promise { + const retryConfig = this.getRetryConfig(step); + const timeout = this.getTimeout(step); + let lastError: Error | undefined; + + for (let attempt = 0; attempt < retryConfig.maxAttempts; attempt++) { + try { + // Apply timeout if configured + if (timeout) { + return await this.executeStepWithTimeout(stepName, step, context, timeout); + } else { + return await this.executeStep(step, context); + } + } catch (error) { + lastError = error as Error; + + // If this was the last attempt, throw + if (attempt >= retryConfig.maxAttempts - 1) { + throw lastError; + } + + // Report the retry number (1-based: retry 1, retry 2, etc.) + onRetry(attempt + 1); + + // Wait before retry + const delay = this.calculateRetryDelay(attempt, retryConfig); + await this.sleep(delay); + } + } + + // Should never reach here, but just in case + throw lastError ?? new WorkflowExecutionError("Step failed with unknown error"); + } + + /** + * Execute a step with timeout + */ + private async executeStepWithTimeout( + stepName: string, + step: Step, + context: WorkflowContext, + timeout: number + ): Promise { + let timer: ReturnType | undefined; + + const timeoutPromise = new Promise((_, reject) => { + timer = setTimeout(() => { + reject(new StepTimeoutError(stepName, timeout)); + }, timeout); + }); + + try { + const result = await Promise.race([this.executeStep(step, context), timeoutPromise]); + return result; + } finally { + // Always clear the timer to prevent memory leaks + if (timer !== undefined) { + clearTimeout(timer); + } + } + } + + /** + * Execute a single step + */ + private async executeStep(step: Step, context: WorkflowContext): Promise { + switch (step.type) { + case "tool": + return this.executeToolStep(step, context); + case "custom": + return this.executeCustomStep(step, context); + case "external": + return this.executeExternalStep(step, context); + case "parallel": + return this.executeParallelStep(step, context); + case "branch": + return this.executeBranchStep(step, context); + default: + throw new WorkflowExecutionError(`Unknown step type: ${(step as Step).type}`); + } + } + + /** + * Execute a tool step + */ + private async executeToolStep(step: ToolStep, context: WorkflowContext): Promise { + // Get input for the tool + const input = step.config?.mapInput ? step.config.mapInput(context) : context.input; + + // Call the tool using the context's callTool function + return context.callTool(step.toolName, input); + } + + /** + * Execute a custom step + */ + private async executeCustomStep(step: CustomStep, context: WorkflowContext): Promise { + // Apply mapInput if configured + if (step.config?.mapInput) { + // Create modified context with mapped input + const mappedInput = step.config.mapInput(context); + const modifiedContext: WorkflowContext = { + ...context, + input: mappedInput, + }; + return step.handler(modifiedContext); + } + + // Execute the custom handler with original context + return step.handler(context); + } + + /** + * Execute an external step + */ + private async executeExternalStep( + step: ExternalStep, + context: WorkflowContext + ): Promise { + try { + // Get input for the tool + const input = step.config?.mapInput ? step.config.mapInput(context) : context.input; + + // Call the external tool + return await context.callExternalTool(step.server, step.toolName, input); + } catch (error) { + // If error is already an ExternalToolError, rethrow it unchanged to avoid double-wrapping + if (error instanceof ExternalToolError) { + throw error; + } + + // Wrap other errors in ExternalToolError with context + throw new ExternalToolError((error as Error).message, step.server, step.toolName, { + originalError: error, + }); + } + } + + /** + * Execute a parallel step + */ + private async executeParallelStep( + step: ParallelStep, + context: WorkflowContext + ): Promise { + // Execute all steps in parallel + const results = await Promise.all( + step.steps.map((childStep) => this.executeStep(childStep, context)) + ); + + // Return array of results + return results; + } + + /** + * Execute a branch step + */ + private async executeBranchStep(step: BranchStep, context: WorkflowContext): Promise { + // Evaluate condition + const conditionResult = await step.condition(context); + + // Execute appropriate branch + const stepsToExecute = conditionResult ? step.thenSteps : (step.elseSteps ?? []); + + // Execute branch steps sequentially + const results: unknown[] = []; + for (const childStep of stepsToExecute) { + const result = await this.executeStep(childStep, context); + results.push(result); + } + + // Return array of results (or undefined if no steps executed) + return results.length > 0 ? results : undefined; + } + + /** + * Get step configuration (all step types have a config property) + */ + private getStepConfig(step: Step): StepConfig | undefined { + return step.config; + } + + /** + * Get retry configuration for a step + */ + private getRetryConfig(step: Step): Required { + const config = this.getStepConfig(step); + + return { + maxAttempts: config?.retry?.maxAttempts ?? 1, + delay: config?.retry?.delay ?? 1000, + backoff: config?.retry?.backoff ?? "linear", + maxDelay: config?.retry?.maxDelay ?? 30000, + }; + } + + /** + * Get error handling strategy for a step + */ + private getErrorHandling(step: Step): ErrorHandling { + return this.getStepConfig(step)?.onError ?? "fail"; + } + + /** + * Get timeout for a step + */ + private getTimeout(step: Step): number | undefined { + return this.getStepConfig(step)?.timeout; + } + + /** + * Calculate retry delay with backoff + */ + private calculateRetryDelay(attempt: number, config: Required): number { + const baseDelay = config.delay; + let delay: number; + + if (config.backoff === "exponential") { + delay = baseDelay * Math.pow(2, attempt); + } else { + // linear + delay = baseDelay * (attempt + 1); + } + + // Cap at maxDelay + return Math.min(delay, config.maxDelay); + } + + /** + * Sleep for specified milliseconds + */ + private sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); + } +} diff --git a/packages/core/src/workflow/external-client.ts b/packages/core/src/workflow/external-client.ts new file mode 100644 index 00000000..cf501c22 --- /dev/null +++ b/packages/core/src/workflow/external-client.ts @@ -0,0 +1,388 @@ +/** + * External MCP tool client + * + * @module workflow/external-client + */ + +import { z } from "zod"; +import { Client } from "@modelcontextprotocol/sdk/client/index.js"; +import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js"; +import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; +import { ExternalToolError } from "./errors"; +import { debugLogger } from "../debug/logger"; + +// ============================================================================= +// INPUT VALIDATION +// ============================================================================= + +/** + * Schema for validating tool input + * Ensures input is a plain object (not null, array, or primitive) + */ +const toolInputSchema = z.record(z.string(), z.unknown()).refine( + (val) => { + // Ensure it's a plain object, not an array + return typeof val === "object" && val !== null && !Array.isArray(val); + }, + { + message: "Tool input must be a plain object (not null, array, or primitive)", + } +); + +// ============================================================================= +// CONNECTION CACHE +// ============================================================================= + +type Transport = StdioClientTransport | StreamableHTTPClientTransport; + +interface CachedConnection { + client: Client; + transport: Transport; + lastUsed: number; +} + +/** + * Configuration options for ExternalToolClient + */ +export interface ExternalToolClientConfig { + /** Cache TTL in milliseconds (default: 5 minutes) */ + cacheTTL?: number; + /** Maximum number of concurrent connections (default: 10) */ + maxConnections?: number; +} + +/** + * External tool client with connection caching + * + * Manages connections to external MCP servers and provides + * a simple interface for calling tools on those servers. + */ +export class ExternalToolClient { + private connections: Map = new Map(); + private pendingConnections: Map> = + new Map(); + private readonly CACHE_TTL: number; + private readonly MAX_CONNECTIONS: number; + private cleanupInterval?: ReturnType; + + /** + * Create a new ExternalToolClient + * + * @param config - Optional configuration for cache behavior + */ + constructor(config: ExternalToolClientConfig = {}) { + this.CACHE_TTL = config.cacheTTL ?? 5 * 60 * 1000; // Default: 5 minutes + this.MAX_CONNECTIONS = config.maxConnections ?? 10; // Default: 10 + + // Start automatic cleanup interval (only in Node.js environments) + // Edge environments should use manual cleanup or shorter-lived clients + if (typeof setInterval !== "undefined" && typeof process !== "undefined") { + // Run cleanup every CACHE_TTL period + this.cleanupInterval = setInterval(() => { + void this.cleanupStaleConnections(); + }, this.CACHE_TTL); + + // Don't prevent process from exiting + if (this.cleanupInterval.unref) { + this.cleanupInterval.unref(); + } + } + } + + /** + * Call a tool on an external MCP server + * + * @param server - MCP server URL or identifier + * @param toolName - Name of the tool to call + * @param input - Input for the tool + * @returns Tool output + * + * @remarks + * This method attempts to extract structured content from the tool response. + * If the response includes `structuredContent`, it is returned directly. + * Otherwise, the method falls back to parsing text content: + * - Locates the first text content item in the response + * - Attempts to parse it as JSON + * - Returns the raw text if JSON parsing fails + * - Returns `undefined` if no content is found + */ + async callTool(server: string, toolName: string, input: unknown): Promise { + // Validate input is a plain object + const validationResult = toolInputSchema.safeParse(input); + if (!validationResult.success) { + throw new ExternalToolError( + `Invalid input for tool "${toolName}" on server "${server}": ${validationResult.error.message}`, + server, + toolName, + { validationError: validationResult.error, providedInput: input } + ); + } + + try { + const client = await this.getOrCreateConnection(server); + + // Call the tool with validated input + const response = await client.callTool( + { + name: toolName, + arguments: validationResult.data, + }, + undefined + ); + + // Extract structured content from response + if (response.structuredContent) { + return response.structuredContent; + } + + // Fallback to parsing text content if no structured content + if (response.content && Array.isArray(response.content) && response.content.length > 0) { + const textContent = response.content.find( + (c: unknown) => typeof c === "object" && c !== null && "type" in c && c.type === "text" + ) as { text?: string } | undefined; + + if (textContent && typeof textContent.text === "string") { + try { + return JSON.parse(textContent.text); + } catch { + return textContent.text; + } + } + } + + return undefined; + } catch (error) { + // If error is already an ExternalToolError, rethrow it unchanged to avoid double-wrapping + if (error instanceof ExternalToolError) { + throw error; + } + + // Wrap other errors in ExternalToolError with context + throw new ExternalToolError( + `Failed to call tool "${toolName}" on server "${server}": ${(error as Error).message}`, + server, + toolName, + { originalError: error } + ); + } + } + + /** + * Get or create a connection to an external server + * + * Handles race conditions: if multiple concurrent calls request the same server, + * they will share the same connection promise to prevent creating duplicate connections. + */ + private async getOrCreateConnection(server: string): Promise { + // Check if we have a cached connection + const cached = this.connections.get(server); + if (cached) { + // Update last used timestamp + cached.lastUsed = Date.now(); + return cached.client; + } + + // Atomically get or create pending connection to prevent race conditions + let pending = this.pendingConnections.get(server); + if (!pending) { + // Evict old connections if cache is full (before creating new promise) + if (this.connections.size >= this.MAX_CONNECTIONS) { + await this.evictOldConnections(); + } + + // Create new connection promise and atomically store it + pending = this.createConnectionWithTransport(server); + this.pendingConnections.set(server, pending); + + // Set up cleanup in a non-blocking way + void pending + .then(({ client, transport }) => { + // Cache the connection with the actual transport for cleanup + this.connections.set(server, { + client, + transport, + lastUsed: Date.now(), + }); + }) + .catch(() => { + // Error will be thrown to awaiting callers, just clean up here + }) + .finally(() => { + // Always remove from pending, whether success or failure + this.pendingConnections.delete(server); + }); + } + + // Wait for the connection (either existing or newly created) + const { client } = await pending; + + // Update timestamp for concurrent access to prevent premature eviction + const nowCached = this.connections.get(server); + if (nowCached) { + nowCached.lastUsed = Date.now(); + } + + return client; + } + + /** + * Create a new MCP client connection with transport + * + * Supports both stdio and HTTP transports: + * - stdio: mcp://server-name or server-name (command in PATH) + * - HTTP: http://... or https://... + * + * @returns Both the client and transport for proper lifecycle management + */ + private async createConnectionWithTransport( + server: string + ): Promise<{ client: Client; transport: Transport }> { + let transport: Transport; + + // Determine transport type based on server identifier + if (server.startsWith("http://") || server.startsWith("https://")) { + // HTTP transport for remote servers + transport = new StreamableHTTPClientTransport(new URL(server)); + } else { + // Stdio transport for local MCP servers + // Parse server identifier (remove mcp:// prefix if present) + const serverName = server.replace(/^mcp:\/\//, ""); + + transport = new StdioClientTransport({ + command: serverName, + args: [], + }); + } + + // Create client + const client = new Client( + { + name: "workflow-client", + version: "1.0.0", + }, + { + capabilities: {}, + } + ); + + // Connect to server + await client.connect(transport); + + return { client, transport }; + } + + /** + * Clean up stale connections that haven't been used within the TTL + * + * Called automatically by the cleanup interval (in Node.js environments) + * or can be called manually for explicit cache management. + */ + private async cleanupStaleConnections(): Promise { + const now = Date.now(); + const toEvict: string[] = []; + + // Find connections older than TTL + for (const [server, connection] of this.connections.entries()) { + if (now - connection.lastUsed > this.CACHE_TTL) { + toEvict.push(server); + } + } + + // Evict stale connections + if (toEvict.length > 0) { + debugLogger.debug("Cleaning up stale MCP connections", { + count: toEvict.length, + servers: toEvict, + }); + + for (const server of toEvict) { + const connection = this.connections.get(server); + if (connection) { + await this.closeConnection(connection, server); + this.connections.delete(server); + } + } + } + } + + /** + * Evict old connections from cache to make room for new ones + */ + private async evictOldConnections(): Promise { + const now = Date.now(); + const toEvict: string[] = []; + + // Find connections older than TTL + for (const [server, connection] of this.connections.entries()) { + if (now - connection.lastUsed > this.CACHE_TTL) { + toEvict.push(server); + } + } + + // If no old connections, evict the least recently used + if (toEvict.length === 0 && this.connections.size > 0) { + let oldestServer: string | undefined; + let oldestTime = Infinity; + + for (const [server, connection] of this.connections.entries()) { + if (connection.lastUsed < oldestTime) { + oldestTime = connection.lastUsed; + oldestServer = server; + } + } + + if (oldestServer) { + toEvict.push(oldestServer); + } + } + + // Evict connections + for (const server of toEvict) { + const connection = this.connections.get(server); + if (connection) { + await this.closeConnection(connection, server); + this.connections.delete(server); + } + } + } + + /** + * Close a connection with fallback transport cleanup + */ + private async closeConnection(connection: CachedConnection, server?: string): Promise { + try { + await connection.client.close(); + } catch (error) { + debugLogger.warn("Failed to close MCP client connection", { error, server }); + + // If Client.close() fails, attempt to close the transport directly + try { + await connection.transport.close(); + } catch (transportError) { + // Log and continue - we've made our best effort at cleanup + debugLogger.error("Failed to close MCP transport", { + error: transportError, + server, + transportType: connection.transport instanceof StdioClientTransport ? "stdio" : "http", + }); + } + } + } + + /** + * Close all connections and stop cleanup interval + */ + async closeAll(): Promise { + // Stop the cleanup interval + if (this.cleanupInterval) { + clearInterval(this.cleanupInterval); + this.cleanupInterval = undefined; + } + + // Close all active connections + for (const [server, connection] of this.connections.entries()) { + await this.closeConnection(connection, server); + } + this.connections.clear(); + } +} diff --git a/packages/core/src/workflow/index.ts b/packages/core/src/workflow/index.ts new file mode 100644 index 00000000..069e88f2 --- /dev/null +++ b/packages/core/src/workflow/index.ts @@ -0,0 +1,276 @@ +/** + * Workflow engine for @mcp-apps-kit/core + * + * Provides a fluent API for composing multi-step workflows as MCP tools. + * Workflows support tool calls, custom logic, parallel execution, conditional + * branching, and configurable error handling. + * + * ## Production Best Practices + * + * ### Automatic Lifecycle Management + * + * Workflows automatically detect your environment and use the appropriate executor manager: + * + * **Traditional Servers** (Node.js, Express, etc.): + * - Uses global singleton with persistent pooling + * - Background cleanup of idle executors (10 min TTL) + * - LRU eviction when pool reaches 100 executors + * - Graceful shutdown via `server.stop()` + * + * **Edge/Serverless** (Supabase, Vercel, Cloudflare, AWS Lambda): + * - Creates fresh manager per invocation (no singleton) + * - Smaller pool size (10 executors, memory-constrained) + * - No background timers (function terminates quickly) + * - Auto-cleanup on function exit via process handlers + * + * Environment detection is automatic based on runtime characteristics. + * + * ### Edge Function Example (Supabase) + * + * ```typescript + * import { serve } from "https://deno.land/std/http/server.ts"; + * import { createApp, workflow, toolStep } from "@mcp-apps-kit/core"; + * + * const myWorkflow = workflow("process_data") + * .describe("Process data") + * .input({ data: z.string() }) + * .step("validate", toolStep("validate")) + * .step("process", toolStep("process")) + * .build(); + * + * const app = createApp({ + * name: "edge-app", + * tools: { myWorkflow }, + * }); + * + * serve(async (req) => { + * // Workflows automatically use edge-optimized manager + * // Cleanup happens when function terminates + * return await app.handleRequest(req); + * }); + * ``` + * + * ### Traditional Server Example + * + * ```typescript + * const server = createApp({ + * name: "my-app", + * tools: { myWorkflow }, + * }); + * + * await server.start(); // Starts background cleanup timer + * + * // Later, during shutdown + * await server.stop(); // Automatically cleans up all workflow resources + * ``` + * + * ### Advanced Configuration + * + * **Traditional Servers:** + * ```typescript + * import { ExecutorManager } from "@mcp-apps-kit/core"; + * + * // Configure the global executor manager + * const manager = ExecutorManager.getInstance({ + * maxExecutors: 200, // Increase pool size for high-traffic apps + * executorTTL: 5 * 60 * 1000, // Cleanup after 5 minutes of inactivity + * autoCleanup: true, // Enable automatic cleanup (default) + * cleanupInterval: 60 * 1000, // Run cleanup every minute + * }); + * + * // Get statistics + * const stats = manager.getStats(); + * console.log(`Active workflows: ${stats.activeExecutors}`); + * console.log(`Total cached: ${stats.totalExecutors}`); + * ``` + * + * **Edge Functions:** + * ```typescript + * import { EdgeExecutorManager } from "@mcp-apps-kit/core"; + * + * // Configure defaults for all edge invocations + * EdgeExecutorManager.configureDefaults({ + * maxExecutors: 5, // Smaller pool for memory-constrained edge + * autoCleanup: false, // No background timers needed + * }); + * ``` + * + * ### External MCP Connection Configuration + * + * For workflows calling external MCP servers, configure connection pooling: + * + * ```typescript + * import { ExternalToolClient } from "@mcp-apps-kit/core"; + * + * // Configure per-executor connection settings + * const client = new ExternalToolClient({ + * cacheTTL: 10 * 60 * 1000, // Keep connections alive for 10 minutes + * maxConnections: 20, // Maximum concurrent MCP connections + * }); + * ``` + * + * Note: Each WorkflowExecutor has its own ExternalToolClient. In traditional + * servers, the ExecutorManager reuses executors efficiently, so connection + * pooling is shared across invocations. In edge functions, connections are + * per-invocation and cleaned up when the function terminates. + * + * @example Sequential workflow + * ```typescript + * import { workflow, toolStep, customStep } from "@mcp-apps-kit/core"; + * + * const orderWorkflow = workflow("process_order") + * .describe("Process a customer order end-to-end") + * .input({ orderId: z.string(), customerId: z.string() }) + * .output({ success: z.boolean(), receiptId: z.string().optional() }) + * .step("validate", toolStep("validate_order")) + * .step("payment", toolStep("process_payment"), { + * retry: { maxAttempts: 3, delay: 1000 }, + * }) + * .build(); + * ``` + * + * @example Parallel execution + * ```typescript + * const notificationsWorkflow = workflow("notifications") + * .describe("Send notifications") + * .input({ userId: z.string() }) + * .parallel("notify", [ + * toolStep("send_email"), + * toolStep("send_sms"), + * toolStep("log_event"), + * ]) + * .build(); + * ``` + * + * @example Conditional branching + * ```typescript + * const shippingWorkflow = workflow("shipping") + * .describe("Handle shipping") + * .input({ orderId: z.string() }) + * .step("validate", toolStep("validate_order")) + * .branch("shipping_method", { + * when: (ctx) => ctx.outputs.validate.isDigital, + * then: [customStep(async (ctx) => ({ delivered: true }))], + * else: [toolStep("create_shipment"), toolStep("notify_warehouse")], + * }) + * .build(); + * ``` + * + * @example External MCP tool + * ```typescript + * const weatherWorkflow = workflow("weather_plan") + * .describe("Plan travel with weather") + * .input({ destination: z.string(), date: z.string() }) + * .step("weather", externalStep({ + * server: "mcp://weather-service", + * tool: "get_forecast", + * mapInput: (ctx) => ({ + * location: ctx.input.destination, + * date: ctx.input.date, + * }), + * })) + * .build(); + * ``` + * + * @module workflow + */ + +// ============================================================================= +// TYPE EXPORTS +// ============================================================================= + +export type { + WorkflowContext, + ToolCaller, + ExternalToolCaller, + ToolValidator, + RetryConfig, + ErrorHandler, + ErrorHandling, + StepConfig, + ToolStep, + CustomStep, + ExternalStep, + ParallelStep, + BranchStep, + Step, + NamedStep, + WorkflowDefinition, + StepExecutionResult, + WorkflowExecutionResult, +} from "./types"; + +export type { + WorkflowBuilderInitial, + WorkflowBuilderWithDescription, + WorkflowBuilderWithInput, + WorkflowBuilderWithOutput, + WorkflowBuilderWithSteps, + SchemaInput, + NormalizedSchema, +} from "./workflow-builder"; + +// ============================================================================= +// ERROR EXPORTS +// ============================================================================= + +export { + WorkflowError, + WorkflowExecutionError, + StepTimeoutError, + ExternalToolError, + WorkflowValidationError, + WorkflowDefinitionError, + ToolResponseValidationError, +} from "./errors"; + +// ============================================================================= +// STEP HELPER EXPORTS +// ============================================================================= + +export { toolStep, customStep, externalStep } from "./steps"; +export type { ExternalStepConfig } from "./steps"; + +// ============================================================================= +// BUILDER FACTORY +// ============================================================================= + +import { WorkflowBuilderImpl } from "./workflow-builder-impl"; +import type { WorkflowBuilderInitial } from "./workflow-builder"; + +/** + * Create a new workflow builder + * + * @param name - Workflow name (will be used as tool name) + * @returns Workflow builder for chaining configuration + * + * @example + * ```typescript + * const myWorkflow = workflow("my_workflow") + * .describe("My workflow description") + * .input({ userId: z.string() }) + * .step("fetch", toolStep("fetch_user")) + * .build(); + * ``` + */ +export function workflow(name: TName): WorkflowBuilderInitial { + return new WorkflowBuilderImpl(name); +} + +// ============================================================================= +// EXECUTOR EXPORT (for advanced use cases) +// ============================================================================= + +export { WorkflowExecutor } from "./executor"; +export { ExternalToolClient } from "./external-client"; +export type { ExternalToolClientConfig } from "./external-client"; + +// ============================================================================= +// EXECUTOR MANAGER EXPORT (production lifecycle management) +// ============================================================================= + +export { ExecutorManager } from "./executor-manager"; +export type { ExecutorManagerConfig } from "./executor-manager"; + +export { EdgeExecutorManager } from "./executor-manager-edge"; +export type { EdgeExecutorManagerConfig } from "./executor-manager-edge"; diff --git a/packages/core/src/workflow/steps.ts b/packages/core/src/workflow/steps.ts new file mode 100644 index 00000000..05b6eb3e --- /dev/null +++ b/packages/core/src/workflow/steps.ts @@ -0,0 +1,133 @@ +/** + * Workflow step helper functions + * + * @module workflow/steps + */ + +import type { ToolStep, CustomStep, ExternalStep, StepConfig, WorkflowContext } from "./types"; + +// ============================================================================= +// TOOL STEP +// ============================================================================= + +/** + * Create a tool step that calls a tool defined in the same app + * + * @param toolName - Name of the tool to call + * @param config - Optional step configuration + * @returns Tool step definition + * + * @example + * ```typescript + * const validateStep = toolStep("validate_order", { + * mapInput: (ctx) => ({ orderId: ctx.input.orderId }), + * retry: { maxAttempts: 3, delay: 1000 }, + * }); + * ``` + */ +export function toolStep( + toolName: string, + config?: StepConfig +): ToolStep { + return { + type: "tool", + toolName, + config, + }; +} + +// ============================================================================= +// CUSTOM STEP +// ============================================================================= + +/** + * Create a custom step with inline handler logic + * + * @param handler - Async function that receives workflow context and returns a value + * @param config - Optional step configuration + * @returns Custom step definition + * + * @example + * ```typescript + * const transformStep = customStep(async (ctx) => { + * const data = ctx.outputs.fetchData; + * return { transformed: data.value * 2 }; + * }, { + * timeout: 5000, + * }); + * ``` + */ +export function customStep( + handler: (context: TContext) => Promise, + config?: StepConfig +): CustomStep { + return { + type: "custom", + handler, + config, + }; +} + +// ============================================================================= +// EXTERNAL STEP +// ============================================================================= + +/** + * Configuration for external tool step + * + * Supports both stdio and HTTP transports: + * - stdio: Server command name (e.g., "weather-server" or "mcp://weather-server") + * - HTTP: Full URL (e.g., "http://localhost:3000/mcp" or "https://api.example.com/mcp") + */ +export interface ExternalStepConfig extends StepConfig { + /** MCP server identifier - command name for stdio or URL for HTTP */ + server: string; + + /** Tool name on the external server */ + tool: string; +} + +/** + * Create an external step that calls a tool from another MCP server + * + * Supports both stdio and HTTP transports based on the server identifier. + * + * @param config - External step configuration including server and tool name + * @returns External step definition + * + * @example stdio transport (local MCP server) + * ```typescript + * const weatherStep = externalStep({ + * server: "weather-server", // or "mcp://weather-server" + * tool: "get_forecast", + * mapInput: (ctx) => ({ + * location: ctx.input.destination, + * date: ctx.input.date, + * }), + * }); + * ``` + * + * @example HTTP transport (remote MCP server) + * ```typescript + * const weatherStep = externalStep({ + * server: "http://localhost:3000/mcp", + * tool: "get_forecast", + * mapInput: (ctx) => ({ + * location: ctx.input.destination, + * date: ctx.input.date, + * }), + * }); + * ``` + */ +export function externalStep( + config: ExternalStepConfig +): ExternalStep { + const { server, tool, ...stepConfig } = config; + + return { + type: "external", + server, + toolName: tool, + config: stepConfig, + }; +} diff --git a/packages/core/src/workflow/types.ts b/packages/core/src/workflow/types.ts new file mode 100644 index 00000000..77d67c13 --- /dev/null +++ b/packages/core/src/workflow/types.ts @@ -0,0 +1,272 @@ +/** + * Workflow engine type definitions + * + * @module workflow/types + */ + +import type { z } from "zod"; +import type { ToolContext } from "../types/tools"; + +// ============================================================================= +// WORKFLOW CONTEXT +// ============================================================================= + +/** + * Type for a validator function or Zod schema + */ +export type ToolValidator = z.ZodType | ((value: unknown) => T); + +/** + * Function to call a tool defined in the same app + * + * @param toolName - Name of the tool to call + * @param input - Input for the tool + * @param validator - Optional Zod schema or validator function to validate the response + * @returns The tool output, validated if a validator is provided + */ +export type ToolCaller = ( + toolName: string, + input: unknown, + validator?: ToolValidator +) => Promise; + +/** + * Function to call a tool from an external MCP server + * + * @param server - MCP server URL or identifier + * @param toolName - Name of the tool to call + * @param input - Input for the tool + * @param validator - Optional Zod schema or validator function to validate the response + * @returns The tool output, validated if a validator is provided + */ +export type ExternalToolCaller = ( + server: string, + toolName: string, + input: unknown, + validator?: ToolValidator +) => Promise; + +/** + * Workflow context available to all steps + * + * Provides access to: + * - Original workflow input + * - Accumulated outputs from previous steps + * - MCP tool context (locale, auth, etc.) + * - Functions to call other tools + */ +export interface WorkflowContext> { + /** Original workflow input */ + input: TInput; + + /** Accumulated outputs from previous steps (keyed by step name) */ + outputs: TOutputs; + + /** MCP tool context (locale, userAgent, subject, etc.) */ + toolContext: ToolContext; + + /** Function to call a tool defined in the same app */ + callTool: ToolCaller; + + /** Function to call a tool from an external MCP server */ + callExternalTool: ExternalToolCaller; +} + +// ============================================================================= +// RETRY CONFIGURATION +// ============================================================================= + +/** + * Retry configuration for workflow steps + */ +export interface RetryConfig { + /** Maximum number of retry attempts (including the initial attempt) */ + maxAttempts: number; + + /** Delay in milliseconds before first retry (default: 1000) */ + delay?: number; + + /** Backoff strategy for retry delays (default: "linear") */ + backoff?: "linear" | "exponential"; + + /** Maximum delay in milliseconds (caps exponential backoff) */ + maxDelay?: number; +} + +// ============================================================================= +// ERROR HANDLING +// ============================================================================= + +/** + * Error handler function type + * + * @param error - The error that occurred + * @param context - Current workflow context + * @returns Optional recovery value or undefined to skip + */ +export type ErrorHandler = ( + error: Error, + context: TContext +) => Promise; + +/** + * Error handling strategy for workflow steps + * + * - "fail": Fail the entire workflow immediately (default) + * - "skip": Skip this step and continue with undefined output + * - Custom function: Handle error and optionally provide recovery value + */ +export type ErrorHandling = "fail" | "skip" | ErrorHandler; + +// ============================================================================= +// STEP CONFIGURATION +// ============================================================================= + +/** + * Base configuration for all workflow steps + */ +export interface StepConfig { + /** Map workflow context to step input (default: passthrough) */ + mapInput?: (context: TContext) => unknown; + + /** Retry configuration for this step */ + retry?: RetryConfig; + + /** Error handling strategy for this step */ + onError?: ErrorHandling; + + /** Timeout in milliseconds for this step */ + timeout?: number; +} + +// ============================================================================= +// STEP TYPES +// ============================================================================= + +/** + * Tool step - calls a tool defined in the same app + */ +export interface ToolStep { + type: "tool"; + toolName: string; + config?: StepConfig; +} + +/** + * Custom step - executes custom async logic + */ +export interface CustomStep { + type: "custom"; + handler: (context: TContext) => Promise; + config?: StepConfig; +} + +/** + * External step - calls a tool from an external MCP server + */ +export interface ExternalStep { + type: "external"; + server: string; + toolName: string; + config?: StepConfig; +} + +/** + * Parallel step - executes multiple steps concurrently + */ +export interface ParallelStep { + type: "parallel"; + steps: Step[]; + config?: Omit, "mapInput">; // Parallel doesn't map input +} + +/** + * Branch step - conditional execution + */ +export interface BranchStep { + type: "branch"; + condition: (context: TContext) => boolean | Promise; + thenSteps: Step[]; + elseSteps?: Step[]; + config?: Omit, "mapInput">; // Branch doesn't map input +} + +/** + * Union type for all step types + */ +export type Step = + | ToolStep + | CustomStep + | ExternalStep + | ParallelStep + | BranchStep; + +// ============================================================================= +// WORKFLOW DEFINITION +// ============================================================================= + +/** + * Named step in a workflow (step with a name for output tracking) + */ +export interface NamedStep { + name: string; + step: Step; +} + +/** + * Internal workflow definition (accumulated by builder) + */ +export interface WorkflowDefinition< + TInput extends z.ZodType = z.ZodType, + TOutput extends z.ZodType = z.ZodType, +> { + name: string; + description?: string; + inputSchema: TInput; + outputSchema?: TOutput; + steps: NamedStep[]; +} + +// ============================================================================= +// EXECUTION RESULT +// ============================================================================= + +/** + * Result of executing a single step + */ +export interface StepExecutionResult { + /** Step name */ + name: string; + + /** Output value from the step */ + output: unknown; + + /** Execution time in milliseconds */ + duration: number; + + /** Number of retry attempts made */ + retries: number; + + /** Error if step failed (only present if onError is not "fail") */ + error?: Error; + + /** Whether the step was skipped due to error */ + skipped?: boolean; +} + +/** + * Result of executing a workflow + */ +export interface WorkflowExecutionResult { + /** Final output from the workflow */ + output: unknown; + + /** All step execution results */ + stepResults: StepExecutionResult[]; + + /** Total execution time in milliseconds */ + duration: number; + + /** Whether the workflow completed successfully */ + success: boolean; +} diff --git a/packages/core/src/workflow/workflow-builder-impl.ts b/packages/core/src/workflow/workflow-builder-impl.ts new file mode 100644 index 00000000..ec926124 --- /dev/null +++ b/packages/core/src/workflow/workflow-builder-impl.ts @@ -0,0 +1,366 @@ +/** + * Workflow builder implementation + * + * @module workflow/workflow-builder-impl + */ + +import { z } from "zod"; +import type { ToolDef, ToolContext } from "../types/tools"; +import type { UIDef } from "../types/ui"; +import type { + WorkflowBuilderInitial, + WorkflowBuilderWithDescription, + WorkflowBuilderWithInput, + WorkflowBuilderWithOutput, + WorkflowBuilderWithSteps, + SchemaInput, + NormalizedSchema, +} from "./workflow-builder"; +import type { + Step, + StepConfig, + NamedStep, + WorkflowContext, + BranchStep, + ParallelStep, +} from "./types"; +import { isZodSchema } from "../utils/schema"; +import { WorkflowValidationError } from "./errors"; +import { ExecutorManager } from "./executor-manager"; +import { EdgeExecutorManager } from "./executor-manager-edge"; +import { debugLogger } from "../debug/logger"; + +// Type declarations for edge runtime detection +declare const Deno: { env?: { get?: (key: string) => string | undefined } } | undefined; +declare const EdgeRuntime: string | undefined; +declare const WorkerGlobalScope: new () => unknown; +declare const self: unknown; + +// ============================================================================= +// ENVIRONMENT DETECTION +// ============================================================================= + +/** + * Detect if we're running in an edge/serverless environment + * + * Edge environments are characterized by short-lived function invocations, + * limited memory, and no persistent background processes. + */ +function isEdgeEnvironment(): boolean { + // Vercel Edge Runtime + if (typeof EdgeRuntime !== "undefined") return true; + + // Deno-based edge (Supabase Edge, Deno Deploy) + if (typeof Deno !== "undefined") return true; + + // Cloudflare Workers - check specific user agent + if (typeof navigator !== "undefined") { + // Cloudflare Workers have a specific user agent + if ( + typeof navigator === "object" && + navigator !== null && + "userAgent" in navigator && + navigator.userAgent === "Cloudflare-Workers" + ) { + return true; + } + } + + // Web Worker environments (includes Service Workers and Cloudflare Workers) + if (typeof WorkerGlobalScope !== "undefined" && typeof self !== "undefined") { + try { + if (self instanceof WorkerGlobalScope) return true; + } catch { + // instanceof might fail in some environments, continue checking + } + } + + // Serverless environments (AWS Lambda, Google Cloud Functions, Vercel, Netlify) + if (typeof process !== "undefined" && process.env) { + const env = process.env; + if ( + env.AWS_LAMBDA_FUNCTION_NAME !== undefined || + env.FUNCTION_NAME !== undefined || + env.VERCEL !== undefined || + env.NETLIFY !== undefined + ) { + return true; + } + } + + return false; +} + +// Module-level cache for edge executor manager +// In edge environments, this prevents creating multiple managers per invocation +let cachedEdgeManager: EdgeExecutorManager | null = null; + +/** + * Reset the cached edge executor manager + * + * This is primarily for testing purposes. In production edge environments, + * the manager is automatically garbage collected when the function instance terminates. + * + * @internal + */ +export async function resetEdgeExecutorManagerCache(): Promise { + if (cachedEdgeManager) { + try { + await cachedEdgeManager.shutdown(); + } catch (error) { + // Log but don't throw - allow reset to complete + debugLogger.error("Error shutting down cached edge executor manager", { error }); + } finally { + // Always clear the cache even if shutdown failed + cachedEdgeManager = null; + } + } +} + +/** + * Get the appropriate executor manager for the current environment + * + * For edge environments, returns a cached manager to prevent accumulation + * in the global cleanup registry. The cached manager is reused across + * tool invocations within the same function instance. + */ +function getExecutorManagerForEnvironment() { + if (isEdgeEnvironment()) { + // Edge: use cached manager per function instance (prevents accumulation) + // The manager will be garbage collected when the function instance terminates + cachedEdgeManager ??= new EdgeExecutorManager(); + return cachedEdgeManager; + } else { + // Traditional: use global singleton with pooling + return ExecutorManager.getInstance(); + } +} + +// ============================================================================= +// INTERNAL BUILDER CONFIG +// ============================================================================= + +interface WorkflowBuilderConfig { + name: string; + description?: string; + inputSchema?: z.ZodType; + outputSchema?: z.ZodType; + steps: NamedStep[]; + ui?: UIDef; +} + +// ============================================================================= +// HELPER FUNCTIONS +// ============================================================================= + +/** + * Normalize schema input to ZodType + */ +function normalizeSchema(schema: T): NormalizedSchema { + if (isZodSchema(schema)) { + return schema as NormalizedSchema; + } + + return z.object(schema) as NormalizedSchema; +} + +/** + * Validate step name is unique + */ +function validateStepName(steps: NamedStep[], name: string): void { + if (steps.some((s) => s.name === name)) { + throw new WorkflowValidationError(`Duplicate step name: "${name}"`); + } +} + +// ============================================================================= +// WORKFLOW BUILDER IMPLEMENTATION +// ============================================================================= + +export class WorkflowBuilderImpl implements WorkflowBuilderInitial { + private config: WorkflowBuilderConfig; + + constructor(name: TName) { + this.config = { + name, + steps: [], + }; + } + + describe(description: string): WorkflowBuilderWithDescription { + this.config.description = description; + return this as unknown as WorkflowBuilderWithDescription; + } + + input( + schema: TInput + ): WorkflowBuilderWithInput> { + this.config.inputSchema = normalizeSchema(schema); + return this as unknown as WorkflowBuilderWithInput>; + } + + output( + this: WorkflowBuilderWithInput, + schema: TOutput + ): WorkflowBuilderWithOutput> { + if (!(this instanceof WorkflowBuilderImpl)) { + throw new Error("WorkflowBuilder method called with invalid context"); + } + const builder = this as unknown as WorkflowBuilderImpl; + builder.config.outputSchema = normalizeSchema(schema); + return this as unknown as WorkflowBuilderWithOutput>; + } + + step( + name: string, + step: Step, + config?: StepConfig + ): WorkflowBuilderWithSteps { + validateStepName(this.config.steps, name); + + // Merge config into step if provided + const stepWithConfig: Step = + config && step.type !== "parallel" && step.type !== "branch" + ? { ...step, config: { ...step.config, ...config } } + : step; + + this.config.steps.push({ + name, + step: stepWithConfig, + }); + + return this as unknown as WorkflowBuilderWithSteps; + } + + parallel( + name: string, + steps: Step[], + config?: Omit + ): WorkflowBuilderWithSteps { + validateStepName(this.config.steps, name); + + const parallelStep: ParallelStep = { + type: "parallel", + steps, + config, + }; + + this.config.steps.push({ + name, + step: parallelStep, + }); + + return this as unknown as WorkflowBuilderWithSteps; + } + + branch( + name: string, + branchConfig: { + when: (context: WorkflowContext) => boolean | Promise; + then: Step[]; + else?: Step[]; + } + ): WorkflowBuilderWithSteps { + validateStepName(this.config.steps, name); + + const branchStep: BranchStep = { + type: "branch", + condition: branchConfig.when, + thenSteps: branchConfig.then, + elseSteps: branchConfig.else, + }; + + this.config.steps.push({ + name, + step: branchStep, + }); + + return this as unknown as WorkflowBuilderWithSteps; + } + + ui(uiDef: UIDef): WorkflowBuilderWithSteps { + this.config.ui = uiDef; + return this as unknown as WorkflowBuilderWithSteps; + } + + build( + this: + | WorkflowBuilderWithInput + | WorkflowBuilderWithOutput + | WorkflowBuilderWithSteps + ): ToolDef { + if (!(this instanceof WorkflowBuilderImpl)) { + throw new Error("WorkflowBuilder method called with invalid context"); + } + const builder = this as unknown as WorkflowBuilderImpl; + + // Validate required fields + if (!builder.config.description) { + throw new WorkflowValidationError("Workflow requires description"); + } + + if (!builder.config.inputSchema) { + throw new WorkflowValidationError("Workflow requires input schema"); + } + + if (builder.config.steps.length === 0) { + throw new WorkflowValidationError("Workflow requires at least one step"); + } + + // Create workflow definition + const definition = { + name: builder.config.name, + description: builder.config.description, + inputSchema: builder.config.inputSchema, + outputSchema: builder.config.outputSchema, + steps: builder.config.steps, + }; + + // Create tool definition with workflow handler + // The ExecutorManager provides: + // - Executor pooling and reuse across invocations (optimal performance) + // - Automatic cleanup of idle executors (prevents memory leaks) + // - LRU eviction when pool is full (bounded resource usage) + // - Reference counting to prevent cleanup during execution + // - Graceful shutdown hooks (proper cleanup on server stop) + // + // In edge environments, pooling is per-invocation and cleanup + // happens automatically when the function terminates. + const toolDef: ToolDef = { + description: builder.config.description, + title: builder.config.name, + input: builder.config.inputSchema as TInput, + output: builder.config.outputSchema as TOutput, + ui: builder.config.ui, + handler: async (input: z.infer, context: ToolContext) => { + // Get the appropriate executor manager for the environment at invocation time + // - Edge/Serverless: creates new manager per invocation (no singleton) + // - Traditional: uses global singleton with pooling and background cleanup + const executorManager = getExecutorManagerForEnvironment(); + + // Get or create the executor (reused across invocations) + const executor = executorManager.getOrCreate(definition); + + // Mark as in-use to prevent cleanup during execution + executorManager.markInUse(definition.name); + + try { + // Execute workflow + const result = await executor.execute(input, context); + + // Return workflow output with metadata support + return result.output as z.infer & { + _meta?: Record; + _text?: string; + _closeWidget?: boolean; + }; + } finally { + // Mark as idle - allows cleanup after TTL expires + executorManager.markIdle(definition.name); + } + }, + }; + + return toolDef; + } +} diff --git a/packages/core/src/workflow/workflow-builder.ts b/packages/core/src/workflow/workflow-builder.ts new file mode 100644 index 00000000..2c6d04a5 --- /dev/null +++ b/packages/core/src/workflow/workflow-builder.ts @@ -0,0 +1,192 @@ +/** + * Workflow builder interfaces + * + * @module workflow/workflow-builder + */ + +import type { z } from "zod"; +import type { ToolDef } from "../types/tools"; +import type { UIDef } from "../types/ui"; +import type { Step, StepConfig, WorkflowContext } from "./types"; + +// ============================================================================= +// SCHEMA INPUT TYPES +// ============================================================================= + +/** + * Input schema accepted by the builder + */ +export type SchemaInput = z.ZodType | z.ZodRawShape; + +/** + * Normalize raw Zod shapes into Zod objects + */ +export type NormalizedSchema = T extends z.ZodType + ? T + : T extends z.ZodRawShape + ? z.ZodObject + : never; + +// ============================================================================= +// BUILDER INTERFACES +// ============================================================================= + +/** + * Step 1: Initial - requires description + */ +export interface WorkflowBuilderInitial { + /** + * Set the workflow description (required) + */ + describe(description: string): WorkflowBuilderWithDescription; +} + +/** + * Step 2: Has description - requires input + */ +export interface WorkflowBuilderWithDescription { + /** + * Set the input schema (required) + */ + input( + schema: TInput + ): WorkflowBuilderWithInput>; +} + +/** + * Step 3: Has input - can add output, steps, ui, or build + */ +export interface WorkflowBuilderWithInput { + /** + * Set the output schema (optional) + */ + output( + schema: TOutput + ): WorkflowBuilderWithOutput>; + + /** + * Add a sequential step to the workflow + */ + step( + name: string, + step: Step, + config?: StepConfig + ): WorkflowBuilderWithSteps; + + /** + * Add a parallel step group to the workflow + */ + parallel(name: string, steps: Step[], config?: Omit): this; + + /** + * Add a conditional branch to the workflow + */ + branch( + name: string, + config: { + when: (context: WorkflowContext) => boolean | Promise; + then: Step[]; + else?: Step[]; + } + ): this; + + /** + * Add a UI definition to the workflow + */ + ui(uiDef: UIDef): this; + + /** + * Build the workflow into a ToolDef + */ + build(): ToolDef; +} + +/** + * Step 4: Has output - can add steps, ui, or build + */ +export interface WorkflowBuilderWithOutput< + TName extends string, + TInput extends z.ZodType, + TOutput extends z.ZodType, +> { + /** + * Add a sequential step to the workflow + */ + step( + name: string, + step: Step, + config?: StepConfig + ): WorkflowBuilderWithSteps; + + /** + * Add a parallel step group to the workflow + */ + parallel(name: string, steps: Step[], config?: Omit): this; + + /** + * Add a conditional branch to the workflow + */ + branch( + name: string, + config: { + when: (context: WorkflowContext) => boolean | Promise; + then: Step[]; + else?: Step[]; + } + ): this; + + /** + * Add a UI definition to the workflow + */ + ui(uiDef: UIDef): this; + + /** + * Build the workflow into a ToolDef + */ + build(): ToolDef; +} + +/** + * Step 5: Has steps - can add more steps, ui, or build + */ +export interface WorkflowBuilderWithSteps< + TName extends string, + TInput extends z.ZodType, + TOutput extends z.ZodType, +> { + /** + * Add a sequential step to the workflow + */ + step( + name: string, + step: Step, + config?: StepConfig + ): WorkflowBuilderWithSteps; + + /** + * Add a parallel step group to the workflow + */ + parallel(name: string, steps: Step[], config?: Omit): this; + + /** + * Add a conditional branch to the workflow + */ + branch( + name: string, + config: { + when: (context: WorkflowContext) => boolean | Promise; + then: Step[]; + else?: Step[]; + } + ): this; + + /** + * Add a UI definition to the workflow + */ + ui(uiDef: UIDef): this; + + /** + * Build the workflow into a ToolDef + */ + build(): ToolDef; +} diff --git a/packages/core/tests/executor-manager.test.ts b/packages/core/tests/executor-manager.test.ts new file mode 100644 index 00000000..4570de30 --- /dev/null +++ b/packages/core/tests/executor-manager.test.ts @@ -0,0 +1,424 @@ +/** + * ExecutorManager tests - Production lifecycle management + */ + +import { describe, it, expect, beforeEach, afterEach } from "vitest"; +import { ExecutorManager } from "../src/workflow/executor-manager"; +import { WorkflowExecutor } from "../src/workflow/executor"; +import type { WorkflowDefinition } from "../src/workflow/types"; +import { z } from "zod"; + +describe("ExecutorManager", () => { + beforeEach(() => { + // Reset the global instance before each test + ExecutorManager.resetInstance(); + }); + + afterEach(async () => { + // Cleanup after each test + await ExecutorManager.getInstance().shutdown(true); + ExecutorManager.resetInstance(); + }); + + describe("Singleton Pattern", () => { + it("should return the same instance on multiple calls", () => { + const instance1 = ExecutorManager.getInstance(); + const instance2 = ExecutorManager.getInstance(); + + expect(instance1).toBe(instance2); + }); + + it("should create new instance after reset", () => { + const instance1 = ExecutorManager.getInstance(); + ExecutorManager.resetInstance(); + const instance2 = ExecutorManager.getInstance(); + + expect(instance1).not.toBe(instance2); + }); + }); + + describe("Executor Creation and Reuse", () => { + it("should create a new executor for a workflow definition", () => { + const manager = ExecutorManager.getInstance(); + const definition: WorkflowDefinition = { + name: "test_workflow", + description: "Test workflow", + inputSchema: z.object({ value: z.string() }), + steps: [], + }; + + const executor1 = manager.getOrCreate(definition); + expect(executor1).toBeInstanceOf(WorkflowExecutor); + }); + + it("should reuse the same executor for the same workflow name", () => { + const manager = ExecutorManager.getInstance(); + const definition: WorkflowDefinition = { + name: "test_workflow", + description: "Test workflow", + inputSchema: z.object({ value: z.string() }), + steps: [], + }; + + const executor1 = manager.getOrCreate(definition); + const executor2 = manager.getOrCreate(definition); + + expect(executor1).toBe(executor2); + }); + + it("should create different executors for different workflow names", () => { + const manager = ExecutorManager.getInstance(); + const definition1: WorkflowDefinition = { + name: "workflow_1", + description: "First workflow", + inputSchema: z.object({ value: z.string() }), + steps: [], + }; + const definition2: WorkflowDefinition = { + name: "workflow_2", + description: "Second workflow", + inputSchema: z.object({ value: z.string() }), + steps: [], + }; + + const executor1 = manager.getOrCreate(definition1); + const executor2 = manager.getOrCreate(definition2); + + expect(executor1).not.toBe(executor2); + }); + }); + + describe("Reference Counting", () => { + it("should track active invocations", () => { + const manager = ExecutorManager.getInstance(); + const definition: WorkflowDefinition = { + name: "test_workflow", + description: "Test workflow", + inputSchema: z.object({ value: z.string() }), + steps: [], + }; + + manager.getOrCreate(definition); + + let stats = manager.getStats(); + expect(stats.activeExecutors).toBe(0); + + manager.markInUse("test_workflow"); + stats = manager.getStats(); + expect(stats.activeExecutors).toBe(1); + + manager.markIdle("test_workflow"); + stats = manager.getStats(); + expect(stats.activeExecutors).toBe(0); + }); + + it("should handle multiple concurrent invocations", () => { + const manager = ExecutorManager.getInstance(); + const definition: WorkflowDefinition = { + name: "test_workflow", + description: "Test workflow", + inputSchema: z.object({ value: z.string() }), + steps: [], + }; + + manager.getOrCreate(definition); + + manager.markInUse("test_workflow"); + manager.markInUse("test_workflow"); + manager.markInUse("test_workflow"); + + const stats = manager.getStats(); + expect(stats.activeExecutors).toBe(1); // Still one executor, but 3 invocations + + manager.markIdle("test_workflow"); + manager.markIdle("test_workflow"); + manager.markIdle("test_workflow"); + + const statsAfter = manager.getStats(); + expect(statsAfter.activeExecutors).toBe(0); + }); + + it("should not go below zero active invocations", () => { + const manager = ExecutorManager.getInstance(); + const definition: WorkflowDefinition = { + name: "test_workflow", + description: "Test workflow", + inputSchema: z.object({ value: z.string() }), + steps: [], + }; + + manager.getOrCreate(definition); + + // Mark idle without marking in-use first + manager.markIdle("test_workflow"); + manager.markIdle("test_workflow"); + + const stats = manager.getStats(); + expect(stats.activeExecutors).toBe(0); + }); + }); + + describe("Statistics", () => { + it("should return accurate statistics", () => { + const manager = ExecutorManager.getInstance(); + + const definition1: WorkflowDefinition = { + name: "workflow_1", + description: "First workflow", + inputSchema: z.object({ value: z.string() }), + steps: [], + }; + const definition2: WorkflowDefinition = { + name: "workflow_2", + description: "Second workflow", + inputSchema: z.object({ value: z.string() }), + steps: [], + }; + + manager.getOrCreate(definition1); + manager.getOrCreate(definition2); + + manager.markInUse("workflow_1"); + + const stats = manager.getStats(); + expect(stats.totalExecutors).toBe(2); + expect(stats.activeExecutors).toBe(1); + expect(stats.idleExecutors).toBe(1); + }); + + it("should track oldest executor age", async () => { + const manager = ExecutorManager.getInstance(); + const definition: WorkflowDefinition = { + name: "test_workflow", + description: "Test workflow", + inputSchema: z.object({ value: z.string() }), + steps: [], + }; + + manager.getOrCreate(definition); + + // Wait a bit to ensure age is non-zero + await new Promise((resolve) => setTimeout(resolve, 50)); + + const stats = manager.getStats(); + expect(stats.oldestExecutorAge).toBeGreaterThan(0); + }); + }); + + describe("LRU Eviction", () => { + it("should evict least recently used executor when cache is full", () => { + const manager = ExecutorManager.getInstance({ maxExecutors: 2 }); + + const def1: WorkflowDefinition = { + name: "workflow_1", + description: "First", + inputSchema: z.object({}), + steps: [], + }; + const def2: WorkflowDefinition = { + name: "workflow_2", + description: "Second", + inputSchema: z.object({}), + steps: [], + }; + const def3: WorkflowDefinition = { + name: "workflow_3", + description: "Third", + inputSchema: z.object({}), + steps: [], + }; + + manager.getOrCreate(def1); + manager.getOrCreate(def2); + + // Cache is full, next one should evict the oldest + manager.getOrCreate(def3); + + const stats = manager.getStats(); + expect(stats.totalExecutors).toBeLessThanOrEqual(2); + }); + }); + + describe("Automatic Cleanup", () => { + it("should cleanup idle executors past TTL", async () => { + const manager = ExecutorManager.getInstance({ + executorTTL: 100, // 100ms TTL + cleanupInterval: 50, // 50ms cleanup interval + }); + + const definition: WorkflowDefinition = { + name: "test_workflow", + description: "Test workflow", + inputSchema: z.object({ value: z.string() }), + steps: [], + }; + + manager.getOrCreate(definition); + + let stats = manager.getStats(); + expect(stats.totalExecutors).toBe(1); + + // Wait for TTL to expire and cleanup to run + await new Promise((resolve) => setTimeout(resolve, 200)); + + stats = manager.getStats(); + expect(stats.totalExecutors).toBe(0); + }, 10000); + + it("should not cleanup active executors", async () => { + const manager = ExecutorManager.getInstance({ + executorTTL: 100, + cleanupInterval: 50, + }); + + const definition: WorkflowDefinition = { + name: "test_workflow", + description: "Test workflow", + inputSchema: z.object({ value: z.string() }), + steps: [], + }; + + manager.getOrCreate(definition); + manager.markInUse("test_workflow"); + + // Wait for TTL to expire + await new Promise((resolve) => setTimeout(resolve, 200)); + + const stats = manager.getStats(); + // Should still have the executor because it's active + expect(stats.totalExecutors).toBe(1); + + manager.markIdle("test_workflow"); + }, 10000); + }); + + describe("Manual Cleanup", () => { + it("should cleanup idle executors on demand", async () => { + const manager = ExecutorManager.getInstance({ + autoCleanup: false, // Disable automatic cleanup + executorTTL: 50, + }); + + const definition: WorkflowDefinition = { + name: "test_workflow", + description: "Test workflow", + inputSchema: z.object({ value: z.string() }), + steps: [], + }; + + manager.getOrCreate(definition); + + // Wait for TTL to expire + await new Promise((resolve) => setTimeout(resolve, 100)); + + let stats = manager.getStats(); + expect(stats.totalExecutors).toBe(1); // Still there because auto-cleanup is off + + // Manual cleanup + await manager.cleanup(); + + stats = manager.getStats(); + expect(stats.totalExecutors).toBe(0); + }); + }); + + describe("Shutdown", () => { + it("should close all executors on shutdown", async () => { + const manager = ExecutorManager.getInstance(); + + const def1: WorkflowDefinition = { + name: "workflow_1", + description: "First", + inputSchema: z.object({}), + steps: [], + }; + const def2: WorkflowDefinition = { + name: "workflow_2", + description: "Second", + inputSchema: z.object({}), + steps: [], + }; + + manager.getOrCreate(def1); + manager.getOrCreate(def2); + + let stats = manager.getStats(); + expect(stats.totalExecutors).toBe(2); + + await manager.shutdown(); + + stats = manager.getStats(); + expect(stats.totalExecutors).toBe(0); + }); + + it("should not close active executors unless forced", async () => { + const manager = ExecutorManager.getInstance(); + + const definition: WorkflowDefinition = { + name: "test_workflow", + description: "Test workflow", + inputSchema: z.object({}), + steps: [], + }; + + manager.getOrCreate(definition); + manager.markInUse("test_workflow"); + + await manager.shutdown(false); // Don't force + + const stats = manager.getStats(); + // Should still have the executor because it's active + expect(stats.totalExecutors).toBe(1); + + manager.markIdle("test_workflow"); + }); + + it("should close active executors when forced", async () => { + const manager = ExecutorManager.getInstance(); + + const definition: WorkflowDefinition = { + name: "test_workflow", + description: "Test workflow", + inputSchema: z.object({}), + steps: [], + }; + + manager.getOrCreate(definition); + manager.markInUse("test_workflow"); + + await manager.shutdown(true); // Force shutdown + + const stats = manager.getStats(); + expect(stats.totalExecutors).toBe(0); + }); + + it("should prevent creating new executors after shutdown", () => { + const manager = ExecutorManager.getInstance(); + + const definition: WorkflowDefinition = { + name: "test_workflow", + description: "Test workflow", + inputSchema: z.object({}), + steps: [], + }; + + manager.shutdown(true); // Start shutdown (don't await) + + expect(() => manager.getOrCreate(definition)).toThrow("ExecutorManager is shutting down"); + }); + }); + + describe("Configuration", () => { + it("should respect custom configuration", () => { + const manager = ExecutorManager.getInstance({ + maxExecutors: 50, + executorTTL: 5000, + autoCleanup: false, + cleanupInterval: 10000, + }); + + // Configuration is applied internally + expect(manager).toBeInstanceOf(ExecutorManager); + }); + }); +}); diff --git a/packages/core/tests/workflow.test.ts b/packages/core/tests/workflow.test.ts new file mode 100644 index 00000000..238d927c --- /dev/null +++ b/packages/core/tests/workflow.test.ts @@ -0,0 +1,663 @@ +/** + * Workflow engine tests + * + * @module workflow.test + */ + +import { describe, it, expect, vi, beforeEach, onTestFinished } from "vitest"; +import { z } from "zod"; +import { + workflow, + toolStep, + customStep, + externalStep, + WorkflowValidationError, + ExecutorManager, + type WorkflowContext, +} from "../src/workflow"; +import type { ToolContext } from "../src/types/tools"; + +// ============================================================================= +// TEST FIXTURES +// ============================================================================= + +const mockToolContext: ToolContext = { + locale: "en-US", + userAgent: "test-agent", + subject: "test-user", +}; + +// Reset ExecutorManager between tests to prevent cross-test pollution +let testCounter = 0; +beforeEach(() => { + ExecutorManager.resetInstance(); + testCounter++; + + // Use onTestFinished to ensure cleanup happens even if test fails + onTestFinished(async () => { + try { + await ExecutorManager.getInstance().shutdown(true); + ExecutorManager.resetInstance(); + } catch (error) { + // Log but don't fail the test cleanup + console.error("Failed to cleanup ExecutorManager:", error); + } + }); +}); + +// Helper to generate unique workflow names per test +function uniqueWorkflowName(base: string): string { + return `${base}_${testCounter}_${Date.now()}`; +} + +// ============================================================================= +// WORKFLOW BUILDER TESTS +// ============================================================================= + +describe("Workflow Builder", () => { + describe("Basic Configuration", () => { + it("should require description", () => { + expect(() => { + const builder = workflow(uniqueWorkflowName("test")) as any; + builder + .input({ value: z.string() }) + .step( + "step1", + customStep(async () => ({ result: "ok" })) + ) + .build(); + }).toThrow(WorkflowValidationError); + }); + + it("should require input schema", () => { + expect(() => { + const builder = workflow(uniqueWorkflowName("test")).describe("Test workflow") as any; + builder + .step( + "step1", + customStep(async () => ({ result: "ok" })) + ) + .build(); + }).toThrow(WorkflowValidationError); + }); + + it("should require at least one step", () => { + expect(() => { + workflow(uniqueWorkflowName("test")) + .describe("Test workflow") + .input({ value: z.string() }) + .build(); + }).toThrow(WorkflowValidationError); + }); + + it("should build valid workflow", () => { + const wf = workflow(uniqueWorkflowName("test")) + .describe("Test workflow") + .input({ value: z.string() }) + .step( + "step1", + customStep(async () => ({ result: "ok" })) + ) + .build(); + + expect(wf).toBeDefined(); + expect(wf.description).toBe("Test workflow"); + expect(wf.handler).toBeDefined(); + }); + + it("should accept output schema", () => { + const wf = workflow(uniqueWorkflowName("test")) + .describe("Test workflow") + .input({ value: z.string() }) + .output({ result: z.string() }) + .step( + "step1", + customStep(async () => ({ result: "ok" })) + ) + .build(); + + expect(wf).toBeDefined(); + expect(wf.output).toBeDefined(); + }); + + it("should reject duplicate step names", () => { + expect(() => { + workflow(uniqueWorkflowName("test")) + .describe("Test workflow") + .input({ value: z.string() }) + .step( + "step1", + customStep(async () => ({ result: "ok" })) + ) + .step( + "step1", + customStep(async () => ({ result: "ok" })) + ) + .build(); + }).toThrow(WorkflowValidationError); + }); + }); + + describe("Step Types", () => { + it("should accept tool step", () => { + const wf = workflow(uniqueWorkflowName("test")) + .describe("Test workflow") + .input({ value: z.string() }) + .step("step1", toolStep("my_tool")) + .build(); + + expect(wf).toBeDefined(); + }); + + it("should accept custom step", () => { + const wf = workflow(uniqueWorkflowName("test")) + .describe("Test workflow") + .input({ value: z.string() }) + .step( + "step1", + customStep(async (ctx) => ({ value: (ctx.input as { value: string }).value })) + ) + .build(); + + expect(wf).toBeDefined(); + }); + + it("should accept external step", () => { + const wf = workflow(uniqueWorkflowName("test")) + .describe("Test workflow") + .input({ value: z.string() }) + .step( + "step1", + externalStep({ + server: "mcp://external-server", + tool: "external_tool", + }) + ) + .build(); + + expect(wf).toBeDefined(); + }); + + it("should accept parallel steps", () => { + const wf = workflow(uniqueWorkflowName("test")) + .describe("Test workflow") + .input({ value: z.string() }) + .parallel("parallel1", [ + customStep(async () => ({ a: 1 })), + customStep(async () => ({ b: 2 })), + ]) + .build(); + + expect(wf).toBeDefined(); + }); + + it("should accept branch steps", () => { + const wf = workflow(uniqueWorkflowName("test")) + .describe("Test workflow") + .input({ value: z.string() }) + .branch("branch1", { + when: (ctx) => (ctx.input as { value: string }).value === "test", + then: [customStep(async () => ({ result: "then" }))], + else: [customStep(async () => ({ result: "else" }))], + }) + .build(); + + expect(wf).toBeDefined(); + }); + }); +}); + +// ============================================================================= +// WORKFLOW EXECUTION TESTS +// ============================================================================= + +describe("Workflow Execution", () => { + describe("Custom Steps", () => { + it("should execute single custom step", async () => { + const wf = workflow(uniqueWorkflowName("test")) + .describe("Test workflow") + .input({ value: z.string() }) + .step( + "step1", + customStep(async (ctx: WorkflowContext) => ({ + result: `processed: ${(ctx.input as { value: string }).value}`, + })) + ) + .build(); + + const result = await wf.handler({ value: "test" }, mockToolContext); + + expect(result).toEqual({ + step1: { result: "processed: test" }, + }); + }); + + it("should execute multiple sequential steps", async () => { + const wf = workflow(uniqueWorkflowName("test")) + .describe("Test workflow") + .input({ value: z.number() }) + .step( + "double", + customStep(async (ctx: WorkflowContext) => ({ + value: (ctx.input as { value: number }).value * 2, + })) + ) + .step( + "add10", + customStep(async (ctx: WorkflowContext) => ({ + value: (ctx.outputs.double as { value: number }).value + 10, + })) + ) + .build(); + + const result = await wf.handler({ value: 5 }, mockToolContext); + + expect(result).toEqual({ + double: { value: 10 }, + add10: { value: 20 }, + }); + }); + + it("should provide accumulated outputs to steps", async () => { + const wf = workflow(uniqueWorkflowName("test")) + .describe("Test workflow") + .input({ value: z.number() }) + .step( + "step1", + customStep(async () => ({ a: 1 })) + ) + .step( + "step2", + customStep(async () => ({ b: 2 })) + ) + .step( + "step3", + customStep(async (ctx) => ({ + sum: (ctx.outputs.step1 as { a: number }).a + (ctx.outputs.step2 as { b: number }).b, + })) + ) + .build(); + + const result = await wf.handler({ value: 0 }, mockToolContext); + + expect(result).toEqual({ + step1: { a: 1 }, + step2: { b: 2 }, + step3: { sum: 3 }, + }); + }); + }); + + describe("Parallel Execution", () => { + it("should execute parallel steps concurrently", async () => { + const executionOrder: number[] = []; + + const wf = workflow(uniqueWorkflowName("test")) + .describe("Test workflow") + .input({ value: z.string() }) + .parallel("parallel1", [ + customStep(async () => { + await new Promise((resolve) => setTimeout(resolve, 50)); + executionOrder.push(1); + return { a: 1 }; + }), + customStep(async () => { + await new Promise((resolve) => setTimeout(resolve, 25)); + executionOrder.push(2); + return { b: 2 }; + }), + customStep(async () => { + executionOrder.push(3); + return { c: 3 }; + }), + ]) + .build(); + + const result = await wf.handler({ value: "test" }, mockToolContext); + + // Results should be collected + expect(result).toEqual({ + parallel1: [{ a: 1 }, { b: 2 }, { c: 3 }], + }); + + // Fastest should complete first (3, then 2, then 1) + expect(executionOrder).toEqual([3, 2, 1]); + }); + }); + + describe("Conditional Branching", () => { + it("should execute then branch when condition is true", async () => { + const wf = workflow(uniqueWorkflowName("test")) + .describe("Test workflow") + .input({ shouldTakeThen: z.boolean() }) + .branch("branch1", { + when: (ctx) => (ctx.input as { shouldTakeThen: boolean }).shouldTakeThen, + then: [customStep(async () => ({ result: "then" }))], + else: [customStep(async () => ({ result: "else" }))], + }) + .build(); + + const result = await wf.handler({ shouldTakeThen: true }, mockToolContext); + + expect(result).toEqual({ + branch1: [{ result: "then" }], + }); + }); + + it("should execute else branch when condition is false", async () => { + const wf = workflow(uniqueWorkflowName("test")) + .describe("Test workflow") + .input({ shouldTakeThen: z.boolean() }) + .branch("branch1", { + when: (ctx) => (ctx.input as { shouldTakeThen: boolean }).shouldTakeThen, + then: [customStep(async () => ({ result: "then" }))], + else: [customStep(async () => ({ result: "else" }))], + }) + .build(); + + const result = await wf.handler({ shouldTakeThen: false }, mockToolContext); + + expect(result).toEqual({ + branch1: [{ result: "else" }], + }); + }); + + it("should handle async condition", async () => { + const wf = workflow(uniqueWorkflowName("test")) + .describe("Test workflow") + .input({ value: z.number() }) + .branch("branch1", { + when: async (ctx) => { + await new Promise((resolve) => setTimeout(resolve, 10)); + return (ctx.input as { value: number }).value > 5; + }, + then: [customStep(async () => ({ result: "high" }))], + else: [customStep(async () => ({ result: "low" }))], + }) + .build(); + + const result1 = await wf.handler({ value: 10 }, mockToolContext); + expect(result1).toEqual({ branch1: [{ result: "high" }] }); + + const result2 = await wf.handler({ value: 3 }, mockToolContext); + expect(result2).toEqual({ branch1: [{ result: "low" }] }); + }); + }); + + describe("Error Handling", () => { + it("should fail workflow by default on step error", async () => { + const wf = workflow(uniqueWorkflowName("test")) + .describe("Test workflow") + .input({ value: z.string() }) + .step( + "failing", + customStep(async () => { + throw new Error("Step failed"); + }) + ) + .build(); + + await expect(wf.handler({ value: "test" }, mockToolContext)).rejects.toThrow(); + }); + + it("should skip step when onError is skip", async () => { + const wf = workflow(uniqueWorkflowName("test")) + .describe("Test workflow") + .input({ value: z.string() }) + .step( + "failing", + customStep(async () => { + throw new Error("Step failed"); + }), + { onError: "skip" } + ) + .step( + "next", + customStep(async () => ({ result: "ok" })) + ) + .build(); + + const result = await wf.handler({ value: "test" }, mockToolContext); + + expect(result).toEqual({ + failing: undefined, + next: { result: "ok" }, + }); + }); + + it("should use custom error handler", async () => { + const errorHandler = vi.fn(async (_error, _ctx) => ({ recovered: true })); + + const wf = workflow(uniqueWorkflowName("test")) + .describe("Test workflow") + .input({ value: z.string() }) + .step( + "failing", + customStep(async () => { + throw new Error("Step failed"); + }), + { onError: errorHandler } + ) + .build(); + + const result = await wf.handler({ value: "test" }, mockToolContext); + + expect(errorHandler).toHaveBeenCalled(); + expect(result).toEqual({ + failing: { recovered: true }, + }); + }); + }); + + describe("Retry Logic", () => { + it("should retry failed step with linear backoff", async () => { + let attempts = 0; + + const wf = workflow(uniqueWorkflowName("test")) + .describe("Test workflow") + .input({ value: z.string() }) + .step( + "retrying", + customStep(async () => { + attempts++; + if (attempts < 3) { + throw new Error("Not yet"); + } + return { success: true }; + }), + { + retry: { maxAttempts: 3, delay: 10, backoff: "linear" }, + } + ) + .build(); + + const result = await wf.handler({ value: "test" }, mockToolContext); + + expect(attempts).toBe(3); + expect(result).toEqual({ + retrying: { success: true }, + }); + }); + + it("should retry with exponential backoff", async () => { + let attempts = 0; + const delays: number[] = []; + let lastTime = Date.now(); + + const wf = workflow(uniqueWorkflowName("test")) + .describe("Test workflow") + .input({ value: z.string() }) + .step( + "retrying", + customStep(async () => { + attempts++; + const now = Date.now(); + if (attempts > 1) { + delays.push(now - lastTime); + } + lastTime = now; + + if (attempts < 3) { + throw new Error("Not yet"); + } + return { success: true }; + }), + { + retry: { maxAttempts: 3, delay: 10, backoff: "exponential" }, + } + ) + .build(); + + await wf.handler({ value: "test" }, mockToolContext); + + expect(attempts).toBe(3); + // Exponential backoff: ~10ms, ~20ms + expect(delays.length).toBe(2); + expect(delays[1]).toBeGreaterThan(delays[0]!); + }); + + it("should respect maxDelay for exponential backoff", async () => { + let attempts = 0; + + const wf = workflow(uniqueWorkflowName("test")) + .describe("Test workflow") + .input({ value: z.string() }) + .step( + "retrying", + customStep(async () => { + attempts++; + if (attempts < 5) { + throw new Error("Not yet"); + } + return { success: true }; + }), + { + retry: { + maxAttempts: 5, + delay: 10, + backoff: "exponential", + maxDelay: 50, + }, + } + ) + .build(); + + const startTime = Date.now(); + await wf.handler({ value: "test" }, mockToolContext); + const duration = Date.now() - startTime; + + // With exponential: 10, 20, 40, 80 (capped at 50) + // So total ~= 10 + 20 + 50 + 50 = 130ms + expect(duration).toBeLessThan(200); // Should be capped + }); + }); + + describe("Step Configuration", () => { + it("should map input using mapInput", async () => { + const wf = workflow(uniqueWorkflowName("test")) + .describe("Test workflow") + .input({ a: z.number(), b: z.number() }) + .step( + "multiply", + customStep(async (ctx: WorkflowContext) => ({ + result: (ctx.input as { product: number }).product * 2, + })), + { + mapInput: (ctx) => ({ + product: + (ctx.input as { a: number; b: number }).a * + (ctx.input as { a: number; b: number }).b, + }), + } + ) + .build(); + + const result = await wf.handler({ a: 3, b: 4 }, mockToolContext); + + expect(result).toEqual({ + multiply: { result: 24 }, // (3 * 4) * 2 + }); + }); + + it("should enforce timeout", async () => { + const wf = workflow(uniqueWorkflowName("test")) + .describe("Test workflow") + .input({ value: z.string() }) + .step( + "slow", + customStep(async () => { + await new Promise((resolve) => setTimeout(resolve, 100)); + return { result: "ok" }; + }), + { timeout: 50 } + ) + .build(); + + await expect(wf.handler({ value: "test" }, mockToolContext)).rejects.toThrow(); + }); + }); +}); + +// ============================================================================= +// STEP HELPER TESTS +// ============================================================================= + +describe("Step Helpers", () => { + describe("toolStep", () => { + it("should create tool step", () => { + const step = toolStep("my_tool"); + + expect(step.type).toBe("tool"); + expect(step.toolName).toBe("my_tool"); + }); + + it("should accept configuration", () => { + const step = toolStep("my_tool", { + retry: { maxAttempts: 3 }, + }); + + expect(step.config?.retry?.maxAttempts).toBe(3); + }); + }); + + describe("customStep", () => { + it("should create custom step", () => { + const handler = async () => ({ result: "ok" }); + const step = customStep(handler); + + expect(step.type).toBe("custom"); + expect(step.handler).toBe(handler); + }); + + it("should accept configuration", () => { + const step = customStep(async () => ({ result: "ok" }), { + timeout: 5000, + }); + + expect(step.config?.timeout).toBe(5000); + }); + }); + + describe("externalStep", () => { + it("should create external step", () => { + const step = externalStep({ + server: "mcp://server", + tool: "tool_name", + }); + + expect(step.type).toBe("external"); + expect(step.server).toBe("mcp://server"); + expect(step.toolName).toBe("tool_name"); + }); + + it("should accept configuration", () => { + const step = externalStep({ + server: "mcp://server", + tool: "tool_name", + retry: { maxAttempts: 2 }, + }); + + expect(step.config?.retry?.maxAttempts).toBe(2); + }); + }); +});