diff --git a/apps/cli/bin/app.js b/apps/cli/bin/app.js index af06a71f..172d56ed 100755 --- a/apps/cli/bin/app.js +++ b/apps/cli/bin/app.js @@ -1,4 +1,32 @@ #!/usr/bin/env node +import yargs from 'yargs'; +import { hideBin } from 'yargs/helpers'; +import { app } from '../dist/app.js'; -import { start } from '../dist/x.js'; -start(); \ No newline at end of file +yargs(hideBin(process.argv)) + .command( + "$0", + "Run rowboatx", + (y) => y + .option("agent", { + type: "string", + description: "The agent to run", + default: "copilot", + }) + .option("run_id", { + type: "string", + description: "Continue an existing run", + }) + .option("input", { + type: "string", + description: "The input to the agent", + }), + (argv) => { + app({ + agent: argv.agent, + runId: argv.run_id, + input: argv.input, + }); + } + ) + .parse(); \ No newline at end of file diff --git a/apps/cli/examples/notebooklm-podcast.txt b/apps/cli/examples/notebooklm-podcast.json similarity index 100% rename from apps/cli/examples/notebooklm-podcast.txt rename to apps/cli/examples/notebooklm-podcast.json diff --git a/apps/cli/package-lock.json b/apps/cli/package-lock.json index d276637b..b54512a7 100644 --- a/apps/cli/package-lock.json +++ b/apps/cli/package-lock.json @@ -16,6 +16,7 @@ "ai": "^5.0.78", "json-schema-to-zod": "^2.6.1", "nanoid": "^5.1.6", + "yargs": "^18.0.0", "zod": "^4.1.12" }, "bin": { @@ -372,6 +373,30 @@ "url": "https://github.com/sponsors/epoberezkin" } }, + "node_modules/ansi-regex": { + "version": "6.2.2", + "resolved": "https://registry.npmjs.org/ansi-regex/-/ansi-regex-6.2.2.tgz", + "integrity": "sha512-Bq3SmSpyFHaWjPk8If9yc6svM8c56dB5BAtW4Qbw5jHTwwXXcTLoRMkpDJp6VL0XzlWaCHTXrkFURMYmD0sLqg==", + "license": "MIT", + "engines": { + "node": ">=12" + }, + "funding": { + "url": "https://github.com/chalk/ansi-regex?sponsor=1" + } + }, + "node_modules/ansi-styles": { + "version": "6.2.3", + "resolved": "https://registry.npmjs.org/ansi-styles/-/ansi-styles-6.2.3.tgz", + "integrity": "sha512-4Dj6M28JB+oAH8kFkTLUo+a2jwOFkuqb3yucU0CANcRRUbxS0cP0nZYCGjcc3BNXwRIsUVmDGgzawme7zvJHvg==", + "license": "MIT", + "engines": { + "node": ">=12" + }, + "funding": { + "url": "https://github.com/chalk/ansi-styles?sponsor=1" + } + }, "node_modules/arg": { "version": "4.1.3", "resolved": "https://registry.npmjs.org/arg/-/arg-4.1.3.tgz", @@ -437,6 +462,20 @@ "url": "https://github.com/sponsors/ljharb" } }, + "node_modules/cliui": { + "version": "9.0.1", + "resolved": "https://registry.npmjs.org/cliui/-/cliui-9.0.1.tgz", + "integrity": "sha512-k7ndgKhwoQveBL+/1tqGJYNz097I7WOvwbmmU2AR5+magtbjPWQTS1C5vzGkBC8Ym8UWRzfKUzUUqFLypY4Q+w==", + "license": "ISC", + "dependencies": { + "string-width": "^7.2.0", + "strip-ansi": "^7.1.0", + "wrap-ansi": "^9.0.0" + }, + "engines": { + "node": ">=20" + } + }, "node_modules/content-disposition": { "version": "1.0.0", "resolved": "https://registry.npmjs.org/content-disposition/-/content-disposition-1.0.0.tgz", @@ -566,6 +605,12 @@ "integrity": "sha512-WMwm9LhRUo+WUaRN+vRuETqG89IgZphVSNkdFgeb6sS/E4OrDIN7t48CAewSHXc6C8lefD8KKfr5vY61brQlow==", "license": "MIT" }, + "node_modules/emoji-regex": { + "version": "10.6.0", + "resolved": "https://registry.npmjs.org/emoji-regex/-/emoji-regex-10.6.0.tgz", + "integrity": "sha512-toUI84YS5YmxW219erniWD0CIVOo46xGKColeNQRgOzDorgBi1v4D71/OFzgD9GO2UGKIv1C3Sp8DAn0+j5w7A==", + "license": "MIT" + }, "node_modules/encodeurl": { "version": "2.0.0", "resolved": "https://registry.npmjs.org/encodeurl/-/encodeurl-2.0.0.tgz", @@ -605,6 +650,15 @@ "node": ">= 0.4" } }, + "node_modules/escalade": { + "version": "3.2.0", + "resolved": "https://registry.npmjs.org/escalade/-/escalade-3.2.0.tgz", + "integrity": "sha512-WUj2qlxaQtO4g6Pq5c29GTcWGDyd8itL8zTlipgECz3JesAiiOKotd8JU6otB3PACgG6xkJUyVhboMS+bje/jA==", + "license": "MIT", + "engines": { + "node": ">=6" + } + }, "node_modules/escape-html": { "version": "1.0.3", "resolved": "https://registry.npmjs.org/escape-html/-/escape-html-1.0.3.tgz", @@ -754,6 +808,27 @@ "url": "https://github.com/sponsors/ljharb" } }, + "node_modules/get-caller-file": { + "version": "2.0.5", + "resolved": "https://registry.npmjs.org/get-caller-file/-/get-caller-file-2.0.5.tgz", + "integrity": "sha512-DyFP3BM/3YHTQOCUL/w0OZHR0lpKeGrxotcHWcqNEdnltqFwXVfhEBQ94eIo34AfQpo0rGki4cyIiftY06h2Fg==", + "license": "ISC", + "engines": { + "node": "6.* || 8.* || >= 10.*" + } + }, + "node_modules/get-east-asian-width": { + "version": "1.4.0", + "resolved": "https://registry.npmjs.org/get-east-asian-width/-/get-east-asian-width-1.4.0.tgz", + "integrity": "sha512-QZjmEOC+IT1uk6Rx0sX22V6uHWVwbdbxf1faPqJ1QhLdGgsRGCZoyaQBm/piRdJy/D2um6hM1UP7ZEeQ4EkP+Q==", + "license": "MIT", + "engines": { + "node": ">=18" + }, + "funding": { + "url": "https://github.com/sponsors/sindresorhus" + } + }, "node_modules/get-intrinsic": { "version": "1.3.0", "resolved": "https://registry.npmjs.org/get-intrinsic/-/get-intrinsic-1.3.0.tgz", @@ -1346,6 +1421,38 @@ "node": ">= 0.8" } }, + "node_modules/string-width": { + "version": "7.2.0", + "resolved": "https://registry.npmjs.org/string-width/-/string-width-7.2.0.tgz", + "integrity": "sha512-tsaTIkKW9b4N+AEj+SVA+WhJzV7/zMhcSu78mLKWSk7cXMOSHsBKFWUs0fWwq8QyK3MgJBQRX6Gbi4kYbdvGkQ==", + "license": "MIT", + "dependencies": { + "emoji-regex": "^10.3.0", + "get-east-asian-width": "^1.0.0", + "strip-ansi": "^7.1.0" + }, + "engines": { + "node": ">=18" + }, + "funding": { + "url": "https://github.com/sponsors/sindresorhus" + } + }, + "node_modules/strip-ansi": { + "version": "7.1.2", + "resolved": "https://registry.npmjs.org/strip-ansi/-/strip-ansi-7.1.2.tgz", + "integrity": "sha512-gmBGslpoQJtgnMAvOVqGZpEz9dyoKTCzy2nfz/n8aIFhN/jCE/rCmcxabB6jOOHV+0WNnylOxaxBQPSvcWklhA==", + "license": "MIT", + "dependencies": { + "ansi-regex": "^6.0.1" + }, + "engines": { + "node": ">=12" + }, + "funding": { + "url": "https://github.com/chalk/strip-ansi?sponsor=1" + } + }, "node_modules/toidentifier": { "version": "1.0.1", "resolved": "https://registry.npmjs.org/toidentifier/-/toidentifier-1.0.1.tgz", @@ -1483,12 +1590,64 @@ "node": ">= 8" } }, + "node_modules/wrap-ansi": { + "version": "9.0.2", + "resolved": "https://registry.npmjs.org/wrap-ansi/-/wrap-ansi-9.0.2.tgz", + "integrity": "sha512-42AtmgqjV+X1VpdOfyTGOYRi0/zsoLqtXQckTmqTeybT+BDIbM/Guxo7x3pE2vtpr1ok6xRqM9OpBe+Jyoqyww==", + "license": "MIT", + "dependencies": { + "ansi-styles": "^6.2.1", + "string-width": "^7.0.0", + "strip-ansi": "^7.1.0" + }, + "engines": { + "node": ">=18" + }, + "funding": { + "url": "https://github.com/chalk/wrap-ansi?sponsor=1" + } + }, "node_modules/wrappy": { "version": "1.0.2", "resolved": "https://registry.npmjs.org/wrappy/-/wrappy-1.0.2.tgz", "integrity": "sha512-l4Sp/DRseor9wL6EvV2+TuQn63dMkPjZ/sp9XkghTEbV9KlPS1xUsZ3u7/IQO4wxtcFB4bgpQPRcR3QCvezPcQ==", "license": "ISC" }, + "node_modules/y18n": { + "version": "5.0.8", + "resolved": "https://registry.npmjs.org/y18n/-/y18n-5.0.8.tgz", + "integrity": "sha512-0pfFzegeDWJHJIAmTLRP2DwHjdF5s7jo9tuztdQxAhINCdvS+3nGINqPd00AphqJR/0LhANUS6/+7SCb98YOfA==", + "license": "ISC", + "engines": { + "node": ">=10" + } + }, + "node_modules/yargs": { + "version": "18.0.0", + "resolved": "https://registry.npmjs.org/yargs/-/yargs-18.0.0.tgz", + "integrity": "sha512-4UEqdc2RYGHZc7Doyqkrqiln3p9X2DZVxaGbwhn2pi7MrRagKaOcIKe8L3OxYcbhXLgLFUS3zAYuQjKBQgmuNg==", + "license": "MIT", + "dependencies": { + "cliui": "^9.0.1", + "escalade": "^3.1.1", + "get-caller-file": "^2.0.5", + "string-width": "^7.2.0", + "y18n": "^5.0.5", + "yargs-parser": "^22.0.0" + }, + "engines": { + "node": "^20.19.0 || ^22.12.0 || >=23" + } + }, + "node_modules/yargs-parser": { + "version": "22.0.0", + "resolved": "https://registry.npmjs.org/yargs-parser/-/yargs-parser-22.0.0.tgz", + "integrity": "sha512-rwu/ClNdSMpkSrUb+d6BRsSkLUq1fmfsY6TOpYzTwvwkg1/NRG85KBy3kq++A8LKQwX6lsu+aWad+2khvuXrqw==", + "license": "ISC", + "engines": { + "node": "^20.19.0 || ^22.12.0 || >=23" + } + }, "node_modules/yn": { "version": "3.1.1", "resolved": "https://registry.npmjs.org/yn/-/yn-3.1.1.tgz", diff --git a/apps/cli/package.json b/apps/cli/package.json index fbcf5b66..23991bc6 100644 --- a/apps/cli/package.json +++ b/apps/cli/package.json @@ -32,6 +32,7 @@ "ai": "^5.0.78", "json-schema-to-zod": "^2.6.1", "nanoid": "^5.1.6", + "yargs": "^18.0.0", "zod": "^4.1.12" } } diff --git a/apps/cli/src/app.ts b/apps/cli/src/app.ts index 2945956c..061bdcb2 100644 --- a/apps/cli/src/app.ts +++ b/apps/cli/src/app.ts @@ -1,117 +1,19 @@ -import { executeWorkflow, resumeWorkflow } from "./application/lib/exec-workflow.js"; +import { streamAgent } from "./application/lib/agent.js"; import { StreamRenderer } from "./application/lib/stream-renderer.js"; -import { createInterface } from "node:readline/promises"; -import { stdin as input, stdout as output } from "node:process"; -type ParsedArgs = { - command: "run" | "resume" | "help" | null; - id: string | null; - interactive: boolean; - message: string; -}; - -function parseArgs(argv: string[]): ParsedArgs { - const args = argv.slice(2); - if (args.length === 0) { - return { command: "help", id: null, interactive: true, message: "" }; - } - - let command: ParsedArgs["command"] = null; - let id: string | null = null; - let interactive = true; - const messageParts: string[] = []; - - if (args[0] !== "run" && args[0] !== "resume") { - command = "help"; - return { command, id: null, interactive, message: "" }; - } - command = args[0]; - - for (let i = 1; i < args.length; i++) { - const a = args[i]; - if (a.startsWith("--")) { - if (a === "--no-interactive") { - interactive = false; - } else if (a.startsWith("--interactive")) { - const [, value] = a.split("="); - if (value === undefined) { - interactive = true; - } else { - interactive = value !== "false"; - } - } - continue; - } - if (!id) { - id = a; - continue; - } - messageParts.push(a); - } - - return { command, id, interactive, message: messageParts.join(" ") }; -} - -function printUsage(): void { - console.log([ - "Usage:", - " rowboatx run [message...] [--interactive | --no-interactive]", - " rowboatx resume [message...] [--interactive | --no-interactive]", - "", - "Flags:", - " --interactive Run interactively (default: true)", - " --no-interactive Disable interactive prompts", - ].join("\n")); -} - -async function promptForResumeInput(): Promise { - const rl = createInterface({ input, output }); - try { - const answer = await rl.question("Enter input to resume the run: "); - return answer; - } finally { - rl.close(); - } -} - -async function render(generator: AsyncGenerator): Promise { +export async function app(opts: { + agent: string; + runId?: string; + input?: string; +}) { const renderer = new StreamRenderer(); - for await (const event of generator) { + for await (const event of streamAgent({ + ...opts, + interactive: true, + })) { renderer.render(event); if (event?.type === "error") { process.exitCode = 1; } } -} - -async function main() { - const { command, id, interactive, message } = parseArgs(process.argv); - - if (command === "help" || !command) { - printUsage(); - return; - } - if (!id) { - printUsage(); - process.exitCode = 1; - return; - } - - switch (command) { - case "run": { - const initialInput = message ?? ""; - await render(executeWorkflow(id, initialInput, interactive)); - break; - } - case "resume": { - const resumeInput = message !== "" ? message : (interactive ? await promptForResumeInput() : ""); - await render(resumeWorkflow(id, resumeInput, interactive)); - break; - } - } -} - -main().catch((err) => { - console.error("Failed:", err instanceof Error ? err.message : String(err)); - process.exitCode = 1; -}); \ No newline at end of file +} \ No newline at end of file diff --git a/apps/cli/src/application/assistant/agent.ts b/apps/cli/src/application/assistant/agent.ts new file mode 100644 index 00000000..d06d85f2 --- /dev/null +++ b/apps/cli/src/application/assistant/agent.ts @@ -0,0 +1,20 @@ +import { Agent, ToolAttachment } from "../entities/agent.js"; +import z from "zod"; +import { CopilotInstructions } from "./instructions.js"; +import { BuiltinTools } from "../lib/builtin-tools.js"; + +const tools: Record> = {}; +for (const [name, tool] of Object.entries(BuiltinTools)) { + tools[name] = { + type: "builtin", + name, + }; +} + +export const CopilotAgent: z.infer = { + name: "rowboatx", + description: "Rowboatx copilot", + instructions: CopilotInstructions, + model: "gpt-4.1", + tools, +} \ No newline at end of file diff --git a/apps/cli/src/application/assistant/chat.ts b/apps/cli/src/application/assistant/chat.ts deleted file mode 100644 index 837c5291..00000000 --- a/apps/cli/src/application/assistant/chat.ts +++ /dev/null @@ -1,731 +0,0 @@ -import { streamText, ModelMessage, tool, stepCountIs } from "ai"; -import * as readline from "readline/promises"; -import { stdin as input, stdout as output } from "process"; -import { z } from "zod"; -import * as fs from "fs/promises"; -import * as path from "path"; -import * as os from "os"; -import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js"; -import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; -import { SSEClientTransport } from "@modelcontextprotocol/sdk/client/sse.js"; -import { Client } from "@modelcontextprotocol/sdk/client/index.js"; -import { StreamRenderer } from "../lib/stream-renderer.js"; -import { getProvider } from "../lib/models.js"; -import { ModelConfig } from "../config/config.js"; -import { executeCommand } from "../lib/command-executor.js"; - -const rl = readline.createInterface({ input, output }); - -// Base directory for file operations - dynamically use user's home directory -const BASE_DIR = path.join(os.homedir(), ".rowboat"); - -// Ensure base directory exists -async function ensureBaseDir() { - try { - await fs.access(BASE_DIR); - } catch { - await fs.mkdir(BASE_DIR, { recursive: true }); - console.log(`šŸ“ Created directory: ${BASE_DIR}\n`); - } -} - -// Export the main copilot function -export async function startCopilot() { - // Conversation history - const messages: ModelMessage[] = []; - - console.log("šŸ¤– Rowboat Copilot - Your Intelligent Workflow Assistant"); - console.log(`šŸ“‚ Working directory: ${BASE_DIR}`); - console.log("šŸ’” I can help you create, manage, and understand workflows."); - console.log("Type 'exit' to quit\n"); - - // Initialize base directory - await ensureBaseDir(); - - while (true) { - // Get user input - const userInput = await rl.question("You: "); - - // Exit condition - if (userInput.toLowerCase() === "exit" || userInput.toLowerCase() === "quit") { - console.log("\nšŸ‘‹ Goodbye!"); - break; - } - - // Add user message to history - messages.push({ role: "user", content: userInput }); - - // Stream AI response - process.stdout.write("\nCopilot: "); - - let currentStep = 0; - const provider = getProvider(); - const result = streamText({ - model: provider(ModelConfig.defaults.model), - messages: messages, - system: `You are an intelligent workflow assistant helping users manage their workflows in ${BASE_DIR}. - -WORKFLOW KNOWLEDGE: -- Workflows are JSON files that orchestrate multiple agents -- Agents are JSON files defining AI assistants with specific tools and instructions -- Tools can be built-in functions or MCP (Model Context Protocol) integrations - -NOTE: Comments with // in the formats below are for explanation only - do NOT include them in actual JSON files - -CORRECT WORKFLOW FORMAT: -{ - "name": "workflow_name", // REQUIRED - must match filename - "description": "Description...", // REQUIRED - must be a description of the workflow - "steps": [ // REQUIRED - array of steps - { - "type": "agent", // REQUIRED - always "agent" - "id": "agent_name" // REQUIRED - must match agent filename - }, - { - "type": "agent", - "id": "another_agent_name" - } - ] -} - -CORRECT AGENT FORMAT (with detailed tool structure): -{ - "name": "agent_name", // REQUIRED - must match filename - "description": "What agent does", // REQUIRED - must be a description of the agent - "model": "gpt-4.1", // REQUIRED - model to use - "instructions": "Instructions...", // REQUIRED - agent instructions - "tools": { // OPTIONAL - can be empty {} or omitted - "descriptive_tool_name": { - "type": "mcp", // REQUIRED - always "mcp" for MCP tools - "name": "actual_mcp_tool_name", // REQUIRED - exact tool name from MCP server - "description": "What tool does", // REQUIRED - clear description - "mcpServerName": "server_name", // REQUIRED - name from mcp.json config - "inputSchema": { // REQUIRED - full JSON schema - "type": "object", - "properties": { - "param1": { - "type": "string", - "description": "Description of param" // description is optional but helpful - } - }, - "required": ["param1"] // OPTIONAL - only include if params are required - } - } - } -} - -IMPORTANT NOTES: -- Agent tools need: type, name, description, mcpServerName, and inputSchema (all REQUIRED) -- Tool keys in agents should be descriptive (like "search", "fetch", "analyze") not the exact tool name -- Agents can have empty tools {} if they don't need external tools -- The "required" array in inputSchema is OPTIONAL - only include it if the tool has required parameters -- If all parameters are optional, you can omit the "required" field entirely -- Property descriptions in inputSchema are optional but helpful for clarity -- All other fields marked REQUIRED must always be present - -EXAMPLE 1 - Firecrawl Search Tool (with required params): -{ - "tools": { - "search": { - "type": "mcp", - "name": "firecrawl_search", - "description": "Search the web", - "mcpServerName": "firecrawl", - "inputSchema": { - "type": "object", - "properties": { - "query": {"type": "string", "description": "Search query"}, - "limit": {"type": "number", "description": "Number of results"}, - "sources": { - "type": "array", - "items": { - "type": "object", - "properties": { - "type": {"type": "string", "enum": ["web", "images", "news"]} - }, - "required": ["type"] - } - } - }, - "required": ["query"] - } - } - } -} - -EXAMPLE 2 - ElevenLabs Text-to-Speech (without required array): -{ - "tools": { - "text_to_speech": { - "type": "mcp", - "name": "text_to_speech", - "description": "Generate audio from text", - "mcpServerName": "elevenLabs", - "inputSchema": { - "type": "object", - "properties": { - "text": {"type": "string"} - } - } - } - } -} - -CRITICAL NAMING AND ORGANIZATION RULES: -- Agent filenames MUST match the "name" field in their JSON (e.g., agent_name.json ā†’ "name": "agent_name") -- Workflow filenames MUST match the "name" field in their JSON (e.g., workflow_name.json ā†’ "name": "workflow_name") -- When referencing agents in workflow steps, the "id" field MUST match the agent's name (e.g., {"type": "agent", "id": "agent_name"}) -- All three must be identical: filename, JSON "name" field, and workflow step "id" field -- ALL workflows MUST be placed in the "workflows/" folder (e.g., workflows/workflow_name.json) -- ALL agents MUST be placed in the "agents/" folder (e.g., agents/agent_name.json) -- NEVER create workflows or agents outside these designated folders -- Always maintain this naming and organizational consistency when creating or updating files - -YOUR CAPABILITIES: -1. Explore the directory structure to understand existing workflows/agents -2. Create new workflows and agents following best practices -3. Update existing files intelligently -4. Read and analyze file contents to maintain consistency -5. Suggest improvements and ask clarifying questions when needed -6. Execute shell commands to perform system operations - - Use executeCommand to run bash/shell commands - - Can list files, check system info, run scripts, etc. - - Commands execute in the .rowboat directory by default -7. List and explore MCP (Model Context Protocol) servers and their available tools - - Use listMcpServers to see all configured MCP servers - - Use listMcpTools to see what tools are available in a specific MCP server - - This helps users understand what external integrations they can use in their workflows - -MCP INTEGRATION: -- MCP servers provide external tools that agents can use (e.g., web scraping, database access, APIs) -- MCP configuration is stored in config/mcp.json -- When users ask about available integrations or tools, check MCP servers -- Help users understand which MCP tools they can add to their agents - -DELETION RULES: -- When a user asks to delete a WORKFLOW, you MUST: - 1. First read/analyze the workflow to identify which agents it uses - 2. List those agents to the user - 3. Ask the user if they want to delete those agents as well - 4. Wait for their response before proceeding with any deletions - 5. Only delete what the user confirms -- When a user asks to delete an AGENT, you MUST: - 1. First read/analyze the agent to identify which workflows it is used in - 2. List those workflows to the user - 3. Ask the user if they want to delete/modify those workflows as well - 4. Wait for their response before proceeding with any deletions - 5. Only delete/modify what the user confirms - -COMMUNICATION STYLE: -- Break down complex tasks into clear steps -- Explore existing files/structure before creating new ones -- Explain your reasoning as you work through tasks -- Be proactive in understanding context -- Confirm what you've done and suggest next steps -- Always ask for confirmation before destructive operations!! - -Always use relative paths (no ${BASE_DIR} prefix) when calling tools.`, - - tools: { - exploreDirectory: tool({ - description: 'Recursively explore directory structure to understand existing workflows, agents, and file organization', - inputSchema: z.object({ - subdirectory: z.string().optional().describe('Subdirectory to explore (optional, defaults to root)'), - maxDepth: z.number().optional().describe('Maximum depth to traverse (default: 3)'), - }), - execute: async ({ subdirectory, maxDepth = 3 }) => { - async function explore(dir: string, depth: number = 0): Promise { - if (depth > maxDepth) return null; - - try { - const entries = await fs.readdir(dir, { withFileTypes: true }); - const result: any = { files: [], directories: {} }; - - for (const entry of entries) { - const fullPath = path.join(dir, entry.name); - if (entry.isFile()) { - const ext = path.extname(entry.name); - const size = (await fs.stat(fullPath)).size; - result.files.push({ - name: entry.name, - type: ext || 'no-extension', - size: size, - relativePath: path.relative(BASE_DIR, fullPath), - }); - } else if (entry.isDirectory()) { - result.directories[entry.name] = await explore(fullPath, depth + 1); - } - } - - return result; - } catch (error) { - return { error: error instanceof Error ? error.message : 'Unknown error' }; - } - } - - const dirPath = subdirectory ? path.join(BASE_DIR, subdirectory) : BASE_DIR; - const structure = await explore(dirPath); - - return { - success: true, - basePath: path.relative(BASE_DIR, dirPath) || '.', - structure, - }; - }, - }), - - readFile: tool({ - description: 'Read and parse file contents. For JSON files, provides parsed structure.', - inputSchema: z.object({ - filename: z.string().describe('The name of the file to read (relative to .rowboat directory)'), - }), - execute: async ({ filename }) => { - try { - const filePath = path.join(BASE_DIR, filename); - const content = await fs.readFile(filePath, 'utf-8'); - - let parsed = null; - let fileType = path.extname(filename); - - if (fileType === '.json') { - try { - parsed = JSON.parse(content); - } catch { - parsed = { error: 'Invalid JSON' }; - } - } - - return { - success: true, - filename, - fileType, - content, - parsed, - path: filePath, - size: content.length, - }; - } catch (error) { - return { - success: false, - message: `Failed to read file: ${error instanceof Error ? error.message : 'Unknown error'}`, - }; - } - }, - }), - - createFile: tool({ - description: 'Create a new file with content. Automatically creates parent directories if needed.', - inputSchema: z.object({ - filename: z.string().describe('The name of the file to create (relative to .rowboat directory)'), - content: z.string().describe('The content to write to the file'), - description: z.string().optional().describe('Optional description of why this file is being created'), - }), - execute: async ({ filename, content, description }) => { - try { - const filePath = path.join(BASE_DIR, filename); - const dir = path.dirname(filePath); - - // Ensure directory exists - await fs.mkdir(dir, { recursive: true }); - - // Write file - await fs.writeFile(filePath, content, 'utf-8'); - - return { - success: true, - message: `File '${filename}' created successfully`, - description: description || 'No description provided', - path: filePath, - size: content.length, - }; - } catch (error) { - return { - success: false, - message: `Failed to create file: ${error instanceof Error ? error.message : 'Unknown error'}`, - }; - } - }, - }), - - updateFile: tool({ - description: 'Update or overwrite the contents of an existing file', - inputSchema: z.object({ - filename: z.string().describe('The name of the file to update (relative to .rowboat directory)'), - content: z.string().describe('The new content to write to the file'), - reason: z.string().optional().describe('Optional reason for the update'), - }), - execute: async ({ filename, content, reason }) => { - try { - const filePath = path.join(BASE_DIR, filename); - - // Check if file exists - await fs.access(filePath); - - // Update file - await fs.writeFile(filePath, content, 'utf-8'); - - return { - success: true, - message: `File '${filename}' updated successfully`, - reason: reason || 'No reason provided', - path: filePath, - size: content.length, - }; - } catch (error) { - return { - success: false, - message: `Failed to update file: ${error instanceof Error ? error.message : 'Unknown error'}`, - }; - } - }, - }), - - deleteFile: tool({ - description: 'Delete a file from the .rowboat directory', - inputSchema: z.object({ - filename: z.string().describe('The name of the file to delete (relative to .rowboat directory)'), - }), - execute: async ({ filename }) => { - try { - const filePath = path.join(BASE_DIR, filename); - await fs.unlink(filePath); - - return { - success: true, - message: `File '${filename}' deleted successfully`, - path: filePath, - }; - } catch (error) { - return { - success: false, - message: `Failed to delete file: ${error instanceof Error ? error.message : 'Unknown error'}`, - }; - } - }, - }), - - listFiles: tool({ - description: 'List all files and directories in the .rowboat directory or subdirectory', - inputSchema: z.object({ - subdirectory: z.string().optional().describe('Optional subdirectory to list (relative to .rowboat directory)'), - }), - execute: async ({ subdirectory }) => { - try { - const dirPath = subdirectory ? path.join(BASE_DIR, subdirectory) : BASE_DIR; - const entries = await fs.readdir(dirPath, { withFileTypes: true }); - - const files = entries - .filter(entry => entry.isFile()) - .map(entry => ({ - name: entry.name, - type: path.extname(entry.name) || 'no-extension', - relativePath: path.relative(BASE_DIR, path.join(dirPath, entry.name)), - })); - - const directories = entries - .filter(entry => entry.isDirectory()) - .map(entry => entry.name); - - return { - success: true, - path: dirPath, - relativePath: path.relative(BASE_DIR, dirPath) || '.', - files, - directories, - totalFiles: files.length, - totalDirectories: directories.length, - }; - } catch (error) { - return { - success: false, - message: `Failed to list files: ${error instanceof Error ? error.message : 'Unknown error'}`, - }; - } - }, - }), - - analyzeWorkflow: tool({ - description: 'Read and analyze a workflow file to understand its structure, agents, and dependencies', - inputSchema: z.object({ - workflowName: z.string().describe('Name of the workflow file to analyze (with or without .json extension)'), - }), - execute: async ({ workflowName }) => { - try { - const filename = workflowName.endsWith('.json') ? workflowName : `${workflowName}.json`; - const filePath = path.join(BASE_DIR, 'workflows', filename); - - const content = await fs.readFile(filePath, 'utf-8'); - const workflow = JSON.parse(content); - - // Extract key information - const analysis = { - name: workflow.name, - description: workflow.description || 'No description', - agentCount: workflow.agents ? workflow.agents.length : 0, - agents: workflow.agents || [], - tools: workflow.tools || {}, - structure: workflow, - }; - - return { - success: true, - filePath: path.relative(BASE_DIR, filePath), - analysis, - }; - } catch (error) { - return { - success: false, - message: `Failed to analyze workflow: ${error instanceof Error ? error.message : 'Unknown error'}`, - }; - } - }, - }), - - listMcpServers: tool({ - description: 'List all available MCP servers from the configuration', - inputSchema: z.object({}), - execute: async () => { - try { - const configPath = path.join(BASE_DIR, 'config', 'mcp.json'); - - // Check if config exists - try { - await fs.access(configPath); - } catch { - return { - success: true, - servers: [], - message: 'No MCP servers configured yet', - }; - } - - const content = await fs.readFile(configPath, 'utf-8'); - const config = JSON.parse(content); - - const servers = Object.keys(config.mcpServers || {}).map(name => { - const server = config.mcpServers[name]; - return { - name, - type: 'command' in server ? 'stdio' : 'http', - command: server.command, - url: server.url, - }; - }); - - return { - success: true, - servers, - count: servers.length, - message: `Found ${servers.length} MCP server(s)`, - }; - } catch (error) { - return { - success: false, - message: `Failed to list MCP servers: ${error instanceof Error ? error.message : 'Unknown error'}`, - }; - } - }, - }), - - listMcpTools: tool({ - description: 'List all available tools from a specific MCP server', - inputSchema: z.object({ - serverName: z.string().describe('Name of the MCP server to query'), - }), - execute: async ({ serverName }) => { - try { - const configPath = path.join(BASE_DIR, 'config', 'mcp.json'); - const content = await fs.readFile(configPath, 'utf-8'); - const config = JSON.parse(content); - - const mcpConfig = config.mcpServers[serverName]; - if (!mcpConfig) { - return { - success: false, - message: `MCP server '${serverName}' not found in configuration`, - }; - } - - // Create transport based on config type - let transport; - if ('command' in mcpConfig) { - transport = new StdioClientTransport({ - command: mcpConfig.command, - args: mcpConfig.args || [], - env: mcpConfig.env || {}, - }); - } else { - try { - transport = new StreamableHTTPClientTransport(new URL(mcpConfig.url)); - } catch { - transport = new SSEClientTransport(new URL(mcpConfig.url)); - } - } - - // Create and connect client - const client = new Client({ - name: 'rowboat-copilot', - version: '1.0.0', - }); - - await client.connect(transport); - - // List available tools - const toolsList = await client.listTools(); - - // Close connection - client.close(); - transport.close(); - - const tools = toolsList.tools.map((t: any) => ({ - name: t.name, - description: t.description || 'No description', - inputSchema: t.inputSchema, - })); - - return { - success: true, - serverName, - tools, - count: tools.length, - message: `Found ${tools.length} tool(s) in MCP server '${serverName}'`, - }; - } catch (error) { - return { - success: false, - message: `Failed to list MCP tools: ${error instanceof Error ? error.message : 'Unknown error'}`, - }; - } - }, - }), - - executeCommand: tool({ - description: 'Execute a shell command and return the output. Use this to run bash/shell commands.', - inputSchema: z.object({ - command: z.string().describe('The shell command to execute (e.g., "ls -la", "cat file.txt")'), - cwd: z.string().optional().describe('Working directory to execute the command in (defaults to .rowboat directory)'), - }), - execute: async ({ command, cwd }) => { - try { - const workingDir = cwd ? path.join(BASE_DIR, cwd) : BASE_DIR; - const result = await executeCommand(command, { cwd: workingDir }); - - return { - success: result.exitCode === 0, - stdout: result.stdout, - stderr: result.stderr, - exitCode: result.exitCode, - command, - workingDir, - }; - } catch (error) { - return { - success: false, - message: `Failed to execute command: ${error instanceof Error ? error.message : 'Unknown error'}`, - command, - }; - } - }, - }), - }, - stopWhen: stepCountIs(20), - }); - - // Initialize renderer with workflow-style output - const renderer = new StreamRenderer({ - showHeaders: false, - dimReasoning: true, - jsonIndent: 2, - truncateJsonAt: 500, - }); - - // Stream and collect response using fullStream - let assistantResponse = ""; - const { fullStream } = result; - - for await (const event of fullStream) { - switch (event.type) { - case "reasoning-start": - renderer.render({ - type: "stream-event", - stepId: "copilot", - event: { type: "reasoning-start" } - }); - break; - case "reasoning-delta": - renderer.render({ - type: "stream-event", - stepId: "copilot", - event: { type: "reasoning-delta", delta: event.text } - }); - break; - case "reasoning-end": - renderer.render({ - type: "stream-event", - stepId: "copilot", - event: { type: "reasoning-end" } - }); - break; - case "text-start": - renderer.render({ - type: "stream-event", - stepId: "copilot", - event: { type: "text-start" } - }); - break; - case "text-delta": - renderer.render({ - type: "stream-event", - stepId: "copilot", - event: { type: "text-delta", delta: event.text } - }); - assistantResponse += event.text; - break; - case "text-end": - renderer.render({ - type: "stream-event", - stepId: "copilot", - event: { type: "text-end" } - }); - break; - case "tool-call": - renderer.render({ - type: "stream-event", - stepId: "copilot", - event: { - type: "tool-call", - toolCallId: event.toolCallId, - toolName: event.toolName, - input: 'args' in event ? event.args : event.input - } - }); - break; - case "tool-result": - // Tool results are not directly rendered in copilot mode - break; - case "finish": - renderer.render({ - type: "stream-event", - stepId: "copilot", - event: { - type: "usage", - usage: event.totalUsage - } - }); - break; - } - } - - console.log(); - - // Add assistant response to history - messages.push({ role: "assistant", content: assistantResponse }); - - // Keep only the last 20 messages (10 user + 10 assistant pairs) - if (messages.length > 20) { - messages.splice(0, messages.length - 20); - } - } - - rl.close(); -} diff --git a/apps/cli/src/application/assistant/instructions.ts b/apps/cli/src/application/assistant/instructions.ts new file mode 100644 index 00000000..c7a750b4 --- /dev/null +++ b/apps/cli/src/application/assistant/instructions.ts @@ -0,0 +1,164 @@ +import { WorkDir as BASE_DIR } from "../config/config.js"; + +export const CopilotInstructions = `You are an intelligent workflow assistant helping users manage their workflows in ${BASE_DIR}. + +WORKFLOW KNOWLEDGE: +- Workflows are JSON files that orchestrate multiple agents +- Agents are JSON files defining AI assistants with specific tools and instructions +- Tools can be built-in functions or MCP (Model Context Protocol) integrations + +NOTE: Comments with // in the formats below are for explanation only - do NOT include them in actual JSON files + +CORRECT WORKFLOW FORMAT: +{ + "name": "workflow_name", // REQUIRED - must match filename + "description": "Description...", // REQUIRED - must be a description of the workflow + "steps": [ // REQUIRED - array of steps + { + "type": "agent", // REQUIRED - always "agent" + "id": "agent_name" // REQUIRED - must match agent filename + }, + { + "type": "agent", + "id": "another_agent_name" + } + ] +} + +CORRECT AGENT FORMAT (with detailed tool structure): +{ + "name": "agent_name", // REQUIRED - must match filename + "description": "What agent does", // REQUIRED - must be a description of the agent + "model": "gpt-4.1", // REQUIRED - model to use + "instructions": "Instructions...", // REQUIRED - agent instructions + "tools": { // OPTIONAL - can be empty {} or omitted + "descriptive_tool_name": { + "type": "mcp", // REQUIRED - always "mcp" for MCP tools + "name": "actual_mcp_tool_name", // REQUIRED - exact tool name from MCP server + "description": "What tool does", // REQUIRED - clear description + "mcpServerName": "server_name", // REQUIRED - name from mcp.json config + "inputSchema": { // REQUIRED - full JSON schema + "type": "object", + "properties": { + "param1": { + "type": "string", + "description": "Description of param" // description is optional but helpful + } + }, + "required": ["param1"] // OPTIONAL - only include if params are required + } + } + } +} + +IMPORTANT NOTES: +- Agent tools need: type, name, description, mcpServerName, and inputSchema (all REQUIRED) +- Tool keys in agents should be descriptive (like "search", "fetch", "analyze") not the exact tool name +- Agents can have empty tools {} if they don't need external tools +- The "required" array in inputSchema is OPTIONAL - only include it if the tool has required parameters +- If all parameters are optional, you can omit the "required" field entirely +- Property descriptions in inputSchema are optional but helpful for clarity +- All other fields marked REQUIRED must always be present + +EXAMPLE 1 - Firecrawl Search Tool (with required params): +{ + "tools": { + "search": { + "type": "mcp", + "name": "firecrawl_search", + "description": "Search the web", + "mcpServerName": "firecrawl", + "inputSchema": { + "type": "object", + "properties": { + "query": {"type": "string", "description": "Search query"}, + "limit": {"type": "number", "description": "Number of results"}, + "sources": { + "type": "array", + "items": { + "type": "object", + "properties": { + "type": {"type": "string", "enum": ["web", "images", "news"]} + }, + "required": ["type"] + } + } + }, + "required": ["query"] + } + } + } +} + +EXAMPLE 2 - ElevenLabs Text-to-Speech (without required array): +{ + "tools": { + "text_to_speech": { + "type": "mcp", + "name": "text_to_speech", + "description": "Generate audio from text", + "mcpServerName": "elevenLabs", + "inputSchema": { + "type": "object", + "properties": { + "text": {"type": "string"} + } + } + } + } +} + +CRITICAL NAMING AND ORGANIZATION RULES: +- Agent filenames MUST match the "name" field in their JSON (e.g., agent_name.json ā†’ "name": "agent_name") +- Workflow filenames MUST match the "name" field in their JSON (e.g., workflow_name.json ā†’ "name": "workflow_name") +- When referencing agents in workflow steps, the "id" field MUST match the agent's name (e.g., {"type": "agent", "id": "agent_name"}) +- All three must be identical: filename, JSON "name" field, and workflow step "id" field +- ALL workflows MUST be placed in the "workflows/" folder (e.g., workflows/workflow_name.json) +- ALL agents MUST be placed in the "agents/" folder (e.g., agents/agent_name.json) +- NEVER create workflows or agents outside these designated folders +- Always maintain this naming and organizational consistency when creating or updating files + +YOUR CAPABILITIES: +1. Explore the directory structure to understand existing workflows/agents +2. Create new workflows and agents following best practices +3. Update existing files intelligently +4. Read and analyze file contents to maintain consistency +5. Suggest improvements and ask clarifying questions when needed +6. Execute shell commands to perform system operations + - Use executeCommand to run bash/shell commands + - Can list files, check system info, run scripts, etc. + - Commands execute in the .rowboat directory by default +7. List and explore MCP (Model Context Protocol) servers and their available tools + - Use listMcpServers to see all configured MCP servers + - Use listMcpTools to see what tools are available in a specific MCP server + - This helps users understand what external integrations they can use in their workflows + +MCP INTEGRATION: +- MCP servers provide external tools that agents can use (e.g., web scraping, database access, APIs) +- MCP configuration is stored in config/mcp.json +- When users ask about available integrations or tools, check MCP servers +- Help users understand which MCP tools they can add to their agents + +DELETION RULES: +- When a user asks to delete a WORKFLOW, you MUST: + 1. First read/analyze the workflow to identify which agents it uses + 2. List those agents to the user + 3. Ask the user if they want to delete those agents as well + 4. Wait for their response before proceeding with any deletions + 5. Only delete what the user confirms +- When a user asks to delete an AGENT, you MUST: + 1. First read/analyze the agent to identify which workflows it is used in + 2. List those workflows to the user + 3. Ask the user if they want to delete/modify those workflows as well + 4. Wait for their response before proceeding with any deletions + 5. Only delete/modify what the user confirms + +COMMUNICATION STYLE: +- Break down complex tasks into clear steps +- Explore existing files/structure before creating new ones +- Explain your reasoning as you work through tasks +- Be proactive in understanding context +- Confirm what you've done and suggest next steps +- Always ask for confirmation before destructive operations!! + +Always use relative paths (no ${BASE_DIR} prefix) when calling tools.`; \ No newline at end of file diff --git a/apps/cli/src/application/config/config.ts b/apps/cli/src/application/config/config.ts index f28a03fb..24c0d013 100644 --- a/apps/cli/src/application/config/config.ts +++ b/apps/cli/src/application/config/config.ts @@ -34,7 +34,7 @@ const baseModelConfig: z.infer = { }, defaults: { provider: "openai", - model: "gpt-4.1", + model: "gpt-5", } }; @@ -55,7 +55,6 @@ function ensureModelConfig() { function ensureDirs() { const ensure = (p: string) => { if (!fs.existsSync(p)) fs.mkdirSync(p, { recursive: true }); }; ensure(WorkDir); - ensure(path.join(WorkDir, "workflows")); ensure(path.join(WorkDir, "agents")); ensure(path.join(WorkDir, "config")); ensureMcpConfig(); diff --git a/apps/cli/src/application/entities/agent.ts b/apps/cli/src/application/entities/agent.ts index e2cd52a4..3c74f109 100644 --- a/apps/cli/src/application/entities/agent.ts +++ b/apps/cli/src/application/entities/agent.ts @@ -1,28 +1,28 @@ import { z } from "zod"; -export const BaseAgentTool = z.object({ +export const BaseTool = z.object({ name: z.string(), }); -export const BuiltinAgentTool = BaseAgentTool.extend({ +export const BuiltinTool = BaseTool.extend({ type: z.literal("builtin"), }); -export const McpAgentTool = BaseAgentTool.extend({ +export const McpTool = BaseTool.extend({ type: z.literal("mcp"), description: z.string(), inputSchema: z.any(), mcpServerName: z.string(), }); -export const WorkflowAgentTool = BaseAgentTool.extend({ - type: z.literal("workflow"), +export const AgentAsATool = BaseTool.extend({ + type: z.literal("agent"), }); -export const AgentTool = z.discriminatedUnion("type", [ - BuiltinAgentTool, - McpAgentTool, - WorkflowAgentTool, +export const ToolAttachment = z.discriminatedUnion("type", [ + BuiltinTool, + McpTool, + AgentAsATool, ]); export const Agent = z.object({ @@ -31,5 +31,5 @@ export const Agent = z.object({ model: z.string().optional(), description: z.string(), instructions: z.string(), - tools: z.record(z.string(), AgentTool).optional(), + tools: z.record(z.string(), ToolAttachment).optional(), }); diff --git a/apps/cli/src/application/entities/llm-step-event.ts b/apps/cli/src/application/entities/llm-step-events.ts similarity index 100% rename from apps/cli/src/application/entities/llm-step-event.ts rename to apps/cli/src/application/entities/llm-step-events.ts diff --git a/apps/cli/src/application/entities/workflow-event.ts b/apps/cli/src/application/entities/run-events.ts similarity index 81% rename from apps/cli/src/application/entities/workflow-event.ts rename to apps/cli/src/application/entities/run-events.ts index 2e463836..47f27ce9 100644 --- a/apps/cli/src/application/entities/workflow-event.ts +++ b/apps/cli/src/application/entities/run-events.ts @@ -1,7 +1,7 @@ import { z } from "zod"; -import { LlmStepStreamEvent } from "./llm-step-event.js"; -import { Workflow } from "./workflow.js"; +import { LlmStepStreamEvent } from "./llm-step-events.js"; import { Message } from "./message.js"; +import { Agent } from "./agent.js"; const BaseRunEvent = z.object({ ts: z.iso.datetime().optional(), @@ -10,47 +10,39 @@ const BaseRunEvent = z.object({ export const RunStartEvent = BaseRunEvent.extend({ type: z.literal("start"), runId: z.string(), - workflowId: z.string(), - workflow: Workflow, + agentId: z.string(), + agent: Agent, interactive: z.boolean(), }); export const RunStepStartEvent = BaseRunEvent.extend({ type: z.literal("step-start"), - stepIndex: z.number(), - stepId: z.string(), - stepType: z.enum(["agent", "function"]), }); export const RunStreamEvent = BaseRunEvent.extend({ type: z.literal("stream-event"), - stepId: z.string(), event: LlmStepStreamEvent, }); export const RunMessageEvent = BaseRunEvent.extend({ type: z.literal("message"), - stepId: z.string(), message: Message, }); export const RunToolInvocationEvent = BaseRunEvent.extend({ type: z.literal("tool-invocation"), - stepId: z.string(), toolName: z.string(), input: z.string(), }); export const RunToolResultEvent = BaseRunEvent.extend({ type: z.literal("tool-result"), - stepId: z.string(), toolName: z.string(), result: z.any(), }); export const RunStepEndEvent = BaseRunEvent.extend({ type: z.literal("step-end"), - stepIndex: z.number(), }); export const RunEndEvent = BaseRunEvent.extend({ diff --git a/apps/cli/src/application/entities/workflow.ts b/apps/cli/src/application/entities/workflow.ts deleted file mode 100644 index 804dad3c..00000000 --- a/apps/cli/src/application/entities/workflow.ts +++ /dev/null @@ -1,21 +0,0 @@ -import { z } from "zod"; - -const AgentStep = z.object({ - type: z.literal("agent"), - id: z.string(), -}); - -const FunctionStep = z.object({ - type: z.literal("function"), - id: z.string(), -}); - -export const Step = z.discriminatedUnion("type", [AgentStep, FunctionStep]); - -export const Workflow = z.object({ - name: z.string(), - description: z.string(), - steps: z.array(Step), - createdAt: z.string().optional(), - updatedAt: z.string().optional(), -}); \ No newline at end of file diff --git a/apps/cli/src/application/functions/get_date.ts b/apps/cli/src/application/functions/get_date.ts deleted file mode 100644 index e8561d80..00000000 --- a/apps/cli/src/application/functions/get_date.ts +++ /dev/null @@ -1,22 +0,0 @@ -import { z } from "zod"; -import { Step, StepOutputT } from "../lib/step.js"; -import { AgentTool } from "../entities/agent.js"; - -export class GetDate implements Step { - async* execute(): StepOutputT { - yield { - type: "text-start", - }; - yield { - type: "text-delta", - delta: 'The current date is ' + new Date().toISOString(), - }; - yield { - type: "text-end", - }; - } - - tools(): Record> { - return {}; - } -} \ No newline at end of file diff --git a/apps/cli/src/application/lib/agent.ts b/apps/cli/src/application/lib/agent.ts index e43d2427..4881e2b4 100644 --- a/apps/cli/src/application/lib/agent.ts +++ b/apps/cli/src/application/lib/agent.ts @@ -1,29 +1,22 @@ -import { Message, MessageList } from "../entities/message.js"; -import { z } from "zod"; -import { Step, StepInputT, StepOutputT } from "./step.js"; -import { ModelMessage, stepCountIs, streamText, tool, Tool, ToolSet, jsonSchema } from "ai"; -import { Agent, AgentTool } from "../entities/agent.js"; -import { ModelConfig, WorkDir } from "../config/config.js"; +import { jsonSchema, ModelMessage } from "ai"; import fs from "fs"; import path from "path"; -import { loadWorkflow } from "./utils.js"; +import { ModelConfig, WorkDir } from "../config/config.js"; +import { Agent, ToolAttachment } from "../entities/agent.js"; +import { createInterface, Interface } from "node:readline/promises"; +import { stdin as input, stdout as output } from "node:process"; +import { AssistantContentPart, AssistantMessage, Message, MessageList, ToolCallPart, ToolMessage, UserMessage } from "../entities/message.js"; +import { runIdGenerator } from "./run-id-gen.js"; +import { LanguageModel, stepCountIs, streamText, tool, Tool, ToolSet } from "ai"; +import { z } from "zod"; import { getProvider } from "./models.js"; +import { LlmStepStreamEvent } from "../entities/llm-step-events.js"; +import { execTool } from "./exec-tool.js"; +import { RunEvent } from "../entities/run-events.js"; +import { CopilotAgent } from "../assistant/agent.js"; +import { BuiltinTools } from "./builtin-tools.js"; -const BashTool = tool({ - description: "Run a command in the shell", - inputSchema: z.object({ - command: z.string(), - }), -}); - -const AskHumanTool = tool({ - description: "Ask the human for input", - inputSchema: z.object({ - question: z.string(), - }), -}); - -function mapAgentTool(t: z.infer): Tool { +export async function mapAgentTool(t: z.infer): Promise { switch (t.type) { case "mcp": return tool({ @@ -31,31 +24,136 @@ function mapAgentTool(t: z.infer): Tool { description: t.description, inputSchema: jsonSchema(t.inputSchema), }); - case "workflow": - const workflow = loadWorkflow(t.name); - if (!workflow) { - throw new Error(`Workflow ${t.name} not found`); + case "agent": + const agent = await loadAgent(t.name); + if (!agent) { + throw new Error(`Agent ${t.name} not found`); } return tool({ name: t.name, - description: workflow.description, + description: agent.description, inputSchema: z.object({ message: z.string().describe("The message to send to the workflow"), }), }); case "builtin": - switch (t.name) { - case "bash": - return BashTool; - case "ask-human": - return AskHumanTool; - default: - throw new Error(`Unknown builtin tool: ${t.name}`); + const match = BuiltinTools[t.name]; + if (!match) { + throw new Error(`Unknown builtin tool: ${t.name}`); } + return tool({ + description: match.description, + inputSchema: match.inputSchema, + }); } } -function convertFromMessages(messages: z.infer[]): ModelMessage[] { +export class RunLogger { + private logFile: string; + private fileHandle: fs.WriteStream; + + ensureRunsDir() { + const runsDir = path.join(WorkDir, "runs"); + if (!fs.existsSync(runsDir)) { + fs.mkdirSync(runsDir, { recursive: true }); + } + } + + constructor(runId: string) { + this.ensureRunsDir(); + this.logFile = path.join(WorkDir, "runs", `${runId}.jsonl`); + this.fileHandle = fs.createWriteStream(this.logFile, { + flags: "a", + encoding: "utf8", + }); + } + + log(event: z.infer) { + if (event.type !== "stream-event") { + this.fileHandle.write(JSON.stringify(event) + "\n"); + } + } + + close() { + this.fileHandle.close(); + } +} + +export class LogAndYield { + private logger: RunLogger + + constructor(logger: RunLogger) { + this.logger = logger; + } + + async *logAndYield(event: z.infer): AsyncGenerator, void, unknown> { + const ev = { + ...event, + ts: new Date().toISOString(), + } + this.logger.log(ev); + yield ev; + } +} + +export class StreamStepMessageBuilder { + private parts: z.infer[] = []; + private textBuffer: string = ""; + private reasoningBuffer: string = ""; + + flushBuffers() { + // skip reasoning + // if (this.reasoningBuffer) { + // this.parts.push({ type: "reasoning", text: this.reasoningBuffer }); + // this.reasoningBuffer = ""; + // } + if (this.textBuffer) { + this.parts.push({ type: "text", text: this.textBuffer }); + this.textBuffer = ""; + } + } + + ingest(event: z.infer) { + switch (event.type) { + case "reasoning-start": + case "reasoning-end": + case "text-start": + case "text-end": + this.flushBuffers(); + break; + case "reasoning-delta": + this.reasoningBuffer += event.delta; + break; + case "text-delta": + this.textBuffer += event.delta; + break; + case "tool-call": + this.parts.push({ + type: "tool-call", + toolCallId: event.toolCallId, + toolName: event.toolName, + arguments: event.input, + }); + break; + } + } + + get(): z.infer { + this.flushBuffers(); + return { + role: "assistant", + content: this.parts, + }; + } +} + +export async function loadAgent(id: string): Promise> { + const agentPath = path.join(WorkDir, "agents", `${id}.json`); + const agent = fs.readFileSync(agentPath, "utf8"); + return Agent.parse(JSON.parse(agent)); +} + +export function convertFromMessages(messages: z.infer[]): ModelMessage[] { const result: ModelMessage[] = []; for (const msg of messages) { switch (msg.role) { @@ -119,100 +217,275 @@ function convertFromMessages(messages: z.infer[]): ModelMessage[ return result; } -export class AgentNode implements Step { - private id: string; - private asTool: boolean; - private agent: z.infer; - constructor(id: string, asTool: boolean) { - this.id = id; - this.asTool = asTool; - const agentPath = path.join(WorkDir, "agents", `${id}.json`); - const agent = fs.readFileSync(agentPath, "utf8"); - this.agent = Agent.parse(JSON.parse(agent)); - } +export async function* streamAgent(opts: { + agent: string; + runId?: string; + input?: string; + interactive?: boolean; +}) { + const messages: z.infer = []; - tools(): Record> { - return this.agent.tools ?? {}; - } - - async* execute(input: StepInputT): StepOutputT { - // console.log("\n\n\t>>>>\t\tinput", JSON.stringify(input)); - const tools: ToolSet = {}; - // if (!this.background) { - // tools["ask-human"] = AskHumanTool; - // } - for (const [name, tool] of Object.entries(this.agent.tools ?? {})) { - if (this.asTool && name === "ask-human") { - continue; - } - try { - tools[name] = mapAgentTool(tool); - } catch (error) { - console.error(`Error mapping tool ${name}:`, error); - continue; - } - } - - // console.log("\n\n\t>>>>\t\ttools", JSON.stringify(tools, null, 2)); - - const provider = getProvider(this.agent.provider); - const { fullStream } = streamText({ - model: provider(this.agent.model || ModelConfig.defaults.model), - messages: convertFromMessages(input), - system: this.agent.instructions, - stopWhen: stepCountIs(1), - tools, - }); - - for await (const event of fullStream) { - // console.log("\n\n\t>>>>\t\tstream event", JSON.stringify(event)); - switch (event.type) { - case "reasoning-start": - yield { - type: "reasoning-start", - }; - break; - case "reasoning-delta": - yield { - type: "reasoning-delta", - delta: event.text, - }; - break; - case "reasoning-end": - yield { - type: "reasoning-end", - }; - break; - case "text-start": - yield { - type: "text-start", - }; - break; - case "text-delta": - yield { - type: "text-delta", - delta: event.text, - }; - break; - case "tool-call": - yield { - type: "tool-call", - toolCallId: event.toolCallId, - toolName: event.toolName, - input: event.input, - }; - break; - case "finish": - yield { - type: "usage", - usage: event.totalUsage, - }; - break; - default: - // console.warn("Unknown event type", event); + // load existing and assemble state if required + if (opts.runId) { + console.error("loading run", opts.runId); + let stream: fs.ReadStream | null = null; + let rl: Interface | null = null; + try { + const logFile = path.join(WorkDir, "runs", `${opts.runId}.jsonl`); + stream = fs.createReadStream(logFile, { encoding: "utf8" }); + rl = createInterface({ input: stream, crlfDelay: Infinity }); + for await (const line of rl) { + if (line.trim() === "") { continue; + } + const parsed = JSON.parse(line); + const event = RunEvent.parse(parsed); + switch (event.type) { + case "message": + messages.push(event.message); + break; + } } + } finally { + stream?.close(); } } -} \ No newline at end of file + + // create runId if not present + if (!opts.runId) { + opts.runId = runIdGenerator.next(); + } + + // load agent data + let agent: z.infer | null = null; + if (opts.agent === "copilot") { + agent = CopilotAgent; + } else { + agent = await loadAgent(opts.agent); + } + if (!agent) { + throw new Error("unable to load agent"); + } + + // set up tools + const tools: ToolSet = {}; + for (const [name, tool] of Object.entries(agent.tools ?? {})) { + try { + tools[name] = await mapAgentTool(tool); + } catch (error) { + console.error(`Error mapping tool ${name}:`, error); + continue; + } + } + + // set up + const logger = new RunLogger(opts.runId); + const ly = new LogAndYield(logger); + const provider = getProvider(agent.provider); + const model = provider(agent.model || ModelConfig.defaults.model); + + // get first input if needed + let rl: Interface | null = null; + if (opts.interactive) { + rl = createInterface({ input, output }); + } + if (opts.input) { + const m: z.infer = { + role: "user", + content: opts.input, + }; + messages.push(m); + yield *ly.logAndYield({ + type: "message", + message: m, + }); + } + try { + // loop b/w user and agent + while (true) { + // get input in interactive mode when last message is not user + if (opts.interactive && (messages.length === 0 || messages[messages.length - 1].role !== "user")) { + const input = await rl!.question("You: "); + // Exit condition + if (["q", "quit", "exit"].includes(input.toLowerCase())) { + console.log("\nšŸ‘‹ Goodbye!"); + return; + } + + const m: z.infer = { + role: "user", + content: input, + }; + messages.push(m); + yield* ly.logAndYield({ + type: "message", + message: m, + }); + } + + // inner loop to handle tool calls + while (true) { + // stream agent response and build message + const messageBuilder = new StreamStepMessageBuilder(); + for await (const event of streamLlm( + model, + messages, + agent.instructions, + tools, + )) { + messageBuilder.ingest(event); + yield* ly.logAndYield({ + type: "stream-event", + event: event, + }); + } + + // build and emit final message from agent response + const msg = messageBuilder.get(); + messages.push(msg); + yield* ly.logAndYield({ + type: "message", + message: msg, + }); + + // handle tool calls + const mappedToolCalls: z.infer[] = []; + let msgToolCallParts: z.infer[] = []; + if (msg.content instanceof Array) { + msgToolCallParts = msg.content.filter(part => part.type === "tool-call"); + } + const hasToolCalls = msgToolCallParts.length > 0; + console.log(msgToolCallParts); + + // validate and map tool calls + for (const part of msgToolCallParts) { + const agentTool = tools[part.toolName]; + if (!agentTool) { + throw new Error(`Tool ${part.toolName} not found`); + } + mappedToolCalls.push({ + toolCall: part, + agentTool: agent.tools![part.toolName], + }); + } + + for (const call of mappedToolCalls) { + const { agentTool, toolCall } = call; + yield* ly.logAndYield({ + type: "tool-invocation", + toolName: toolCall.toolName, + input: JSON.stringify(toolCall.arguments), + }); + const result = await execTool(agentTool, toolCall.arguments); + const resultMsg: z.infer = { + role: "tool", + content: JSON.stringify(result), + toolCallId: toolCall.toolCallId, + toolName: toolCall.toolName, + }; + messages.push(resultMsg); + yield* ly.logAndYield({ + type: "tool-result", + toolName: toolCall.toolName, + result: result, + }); + yield* ly.logAndYield({ + type: "message", + message: resultMsg, + }); + } + + // if the agent response had tool calls, replay this agent + if (hasToolCalls) { + continue; + } + + // otherwise, break + break; + } + + // if not interactive, return + if (!opts.interactive) { + break; + } + } + } finally { + rl?.close(); + logger.close(); + } +} + +async function* streamLlm( + model: LanguageModel, + messages: z.infer, + instructions: string, + tools: ToolSet, +): AsyncGenerator, void, unknown> { + const { fullStream } = streamText({ + model, + messages: convertFromMessages(messages), + system: instructions, + tools, + stopWhen: stepCountIs(1), + providerOptions: { + openai: { + reasoningEffort: "low", + reasoningSummary: "auto", + }, + } + }); + for await (const event of fullStream) { + // console.log("\n\n\t>>>>\t\tstream event", JSON.stringify(event)); + switch (event.type) { + case "reasoning-start": + yield { + type: "reasoning-start", + }; + break; + case "reasoning-delta": + yield { + type: "reasoning-delta", + delta: event.text, + }; + break; + case "reasoning-end": + yield { + type: "reasoning-end", + }; + break; + case "text-start": + yield { + type: "text-start", + }; + break; + case "text-delta": + yield { + type: "text-delta", + delta: event.text, + }; + break; + case "tool-call": + yield { + type: "tool-call", + toolCallId: event.toolCallId, + toolName: event.toolName, + input: event.input, + }; + break; + case "finish": + yield { + type: "usage", + usage: event.totalUsage, + }; + break; + default: + // console.warn("Unknown event type", event); + continue; + } + } +} +export const MappedToolCall = z.object({ + toolCall: ToolCallPart, + agentTool: ToolAttachment, +}); diff --git a/apps/cli/src/application/lib/builtin-tools.ts b/apps/cli/src/application/lib/builtin-tools.ts new file mode 100644 index 00000000..9af6a9fe --- /dev/null +++ b/apps/cli/src/application/lib/builtin-tools.ts @@ -0,0 +1,424 @@ +import { z, ZodType } from "zod"; +import * as fs from "fs/promises"; +import * as path from "path"; +import { WorkDir as BASE_DIR } from "../config/config.js"; +import { executeCommand } from "./command-executor.js"; +import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js"; +import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; +import { SSEClientTransport } from "@modelcontextprotocol/sdk/client/sse.js"; +import { Client } from "@modelcontextprotocol/sdk/client"; + +const BuiltinToolsSchema = z.record(z.string(), z.object({ + description: z.string(), + inputSchema: z.custom(), + execute: z.function({ + input: z.any(), + output: z.promise(z.any()), + }), +})); + +export const BuiltinTools: z.infer = { + exploreDirectory: { + description: 'Recursively explore directory structure to understand existing workflows, agents, and file organization', + inputSchema: z.object({ + subdirectory: z.string().optional().describe('Subdirectory to explore (optional, defaults to root)'), + maxDepth: z.number().optional().describe('Maximum depth to traverse (default: 3)'), + }), + execute: async ({ subdirectory, maxDepth = 3 }: { subdirectory?: string, maxDepth?: number }) => { + async function explore(dir: string, depth: number = 0): Promise { + if (depth > maxDepth) return null; + + try { + const entries = await fs.readdir(dir, { withFileTypes: true }); + const result: any = { files: [], directories: {} }; + + for (const entry of entries) { + const fullPath = path.join(dir, entry.name); + if (entry.isFile()) { + const ext = path.extname(entry.name); + const size = (await fs.stat(fullPath)).size; + result.files.push({ + name: entry.name, + type: ext || 'no-extension', + size: size, + relativePath: path.relative(BASE_DIR, fullPath), + }); + } else if (entry.isDirectory()) { + result.directories[entry.name] = await explore(fullPath, depth + 1); + } + } + + return result; + } catch (error) { + return { error: error instanceof Error ? error.message : 'Unknown error' }; + } + } + + const dirPath = subdirectory ? path.join(BASE_DIR, subdirectory) : BASE_DIR; + const structure = await explore(dirPath); + + return { + success: true, + basePath: path.relative(BASE_DIR, dirPath) || '.', + structure, + }; + }, + }, + + readFile: { + description: 'Read and parse file contents. For JSON files, provides parsed structure.', + inputSchema: z.object({ + filename: z.string().describe('The name of the file to read (relative to .rowboat directory)'), + }), + execute: async ({ filename }: { filename: string }) => { + try { + const filePath = path.join(BASE_DIR, filename); + const content = await fs.readFile(filePath, 'utf-8'); + + let parsed = null; + let fileType = path.extname(filename); + + if (fileType === '.json') { + try { + parsed = JSON.parse(content); + } catch { + parsed = { error: 'Invalid JSON' }; + } + } + + return { + success: true, + filename, + fileType, + content, + parsed, + path: filePath, + size: content.length, + }; + } catch (error) { + return { + success: false, + message: `Failed to read file: ${error instanceof Error ? error.message : 'Unknown error'}`, + }; + } + }, + }, + + createFile: { + description: 'Create a new file with content. Automatically creates parent directories if needed.', + inputSchema: z.object({ + filename: z.string().describe('The name of the file to create (relative to .rowboat directory)'), + content: z.string().describe('The content to write to the file'), + description: z.string().optional().describe('Optional description of why this file is being created'), + }), + execute: async ({ filename, content, description }: { filename: string, content: string, description?: string }) => { + try { + const filePath = path.join(BASE_DIR, filename); + const dir = path.dirname(filePath); + + // Ensure directory exists + await fs.mkdir(dir, { recursive: true }); + + // Write file + await fs.writeFile(filePath, content, 'utf-8'); + + return { + success: true, + message: `File '${filename}' created successfully`, + description: description || 'No description provided', + path: filePath, + size: content.length, + }; + } catch (error) { + return { + success: false, + message: `Failed to create file: ${error instanceof Error ? error.message : 'Unknown error'}`, + }; + } + }, + }, + + updateFile: { + description: 'Update or overwrite the contents of an existing file', + inputSchema: z.object({ + filename: z.string().describe('The name of the file to update (relative to .rowboat directory)'), + content: z.string().describe('The new content to write to the file'), + reason: z.string().optional().describe('Optional reason for the update'), + }), + execute: async ({ filename, content, reason }: { filename: string, content: string, reason?: string }) => { + try { + const filePath = path.join(BASE_DIR, filename); + + // Check if file exists + await fs.access(filePath); + + // Update file + await fs.writeFile(filePath, content, 'utf-8'); + + return { + success: true, + message: `File '${filename}' updated successfully`, + reason: reason || 'No reason provided', + path: filePath, + size: content.length, + }; + } catch (error) { + return { + success: false, + message: `Failed to update file: ${error instanceof Error ? error.message : 'Unknown error'}`, + }; + } + }, + }, + + deleteFile: { + description: 'Delete a file from the .rowboat directory', + inputSchema: z.object({ + filename: z.string().describe('The name of the file to delete (relative to .rowboat directory)'), + }), + execute: async ({ filename }: { filename: string }) => { + try { + const filePath = path.join(BASE_DIR, filename); + await fs.unlink(filePath); + + return { + success: true, + message: `File '${filename}' deleted successfully`, + path: filePath, + }; + } catch (error) { + return { + success: false, + message: `Failed to delete file: ${error instanceof Error ? error.message : 'Unknown error'}`, + }; + } + }, + }, + + listFiles: { + description: 'List all files and directories in the .rowboat directory or subdirectory', + inputSchema: z.object({ + subdirectory: z.string().optional().describe('Optional subdirectory to list (relative to .rowboat directory)'), + }), + execute: async ({ subdirectory }: { subdirectory?: string }) => { + try { + const dirPath = subdirectory ? path.join(BASE_DIR, subdirectory) : BASE_DIR; + const entries = await fs.readdir(dirPath, { withFileTypes: true }); + + const files = entries + .filter(entry => entry.isFile()) + .map(entry => ({ + name: entry.name, + type: path.extname(entry.name) || 'no-extension', + relativePath: path.relative(BASE_DIR, path.join(dirPath, entry.name)), + })); + + const directories = entries + .filter(entry => entry.isDirectory()) + .map(entry => entry.name); + + return { + success: true, + path: dirPath, + relativePath: path.relative(BASE_DIR, dirPath) || '.', + files, + directories, + totalFiles: files.length, + totalDirectories: directories.length, + }; + } catch (error) { + return { + success: false, + message: `Failed to list files: ${error instanceof Error ? error.message : 'Unknown error'}`, + }; + } + }, + }, + + analyzeWorkflow: { + description: 'Read and analyze a workflow file to understand its structure, agents, and dependencies', + inputSchema: z.object({ + workflowName: z.string().describe('Name of the workflow file to analyze (with or without .json extension)'), + }), + execute: async ({ workflowName }: { workflowName: string }) => { + try { + const filename = workflowName.endsWith('.json') ? workflowName : `${workflowName}.json`; + const filePath = path.join(BASE_DIR, 'workflows', filename); + + const content = await fs.readFile(filePath, 'utf-8'); + const workflow = JSON.parse(content); + + // Extract key information + const analysis = { + name: workflow.name, + description: workflow.description || 'No description', + agentCount: workflow.agents ? workflow.agents.length : 0, + agents: workflow.agents || [], + tools: workflow.tools || {}, + structure: workflow, + }; + + return { + success: true, + filePath: path.relative(BASE_DIR, filePath), + analysis, + }; + } catch (error) { + return { + success: false, + message: `Failed to analyze workflow: ${error instanceof Error ? error.message : 'Unknown error'}`, + }; + } + }, + }, + + listMcpServers: { + description: 'List all available MCP servers from the configuration', + inputSchema: z.object({}), + execute: async (): Promise<{ success: boolean, servers: any[], count: number, message: string }> => { + try { + const configPath = path.join(BASE_DIR, 'config', 'mcp.json'); + + // Check if config exists + try { + await fs.access(configPath); + } catch { + return { + success: true, + servers: [], + count: 0, + message: 'No MCP servers configured yet', + }; + } + + const content = await fs.readFile(configPath, 'utf-8'); + const config = JSON.parse(content); + + const servers = Object.keys(config.mcpServers || {}).map(name => { + const server = config.mcpServers[name]; + return { + name, + type: 'command' in server ? 'stdio' : 'http', + command: server.command, + url: server.url, + }; + }); + + return { + success: true, + servers, + count: servers.length, + message: `Found ${servers.length} MCP server(s)`, + }; + } catch (error) { + return { + success: false, + servers: [], + count: 0, + message: `Failed to list MCP servers: ${error instanceof Error ? error.message : 'Unknown error'}`, + }; + } + }, + }, + + listMcpTools: { + description: 'List all available tools from a specific MCP server', + inputSchema: z.object({ + serverName: z.string().describe('Name of the MCP server to query'), + }), + execute: async ({ serverName }: { serverName: string }) => { + try { + const configPath = path.join(BASE_DIR, 'config', 'mcp.json'); + const content = await fs.readFile(configPath, 'utf-8'); + const config = JSON.parse(content); + + const mcpConfig = config.mcpServers[serverName]; + if (!mcpConfig) { + return { + success: false, + message: `MCP server '${serverName}' not found in configuration`, + }; + } + + // Create transport based on config type + let transport; + if ('command' in mcpConfig) { + transport = new StdioClientTransport({ + command: mcpConfig.command, + args: mcpConfig.args || [], + env: mcpConfig.env || {}, + }); + } else { + try { + transport = new StreamableHTTPClientTransport(new URL(mcpConfig.url)); + } catch { + transport = new SSEClientTransport(new URL(mcpConfig.url)); + } + } + + // Create and connect client + const client = new Client({ + name: 'rowboat-copilot', + version: '1.0.0', + }); + + await client.connect(transport); + + // List available tools + const toolsList = await client.listTools(); + + // Close connection + client.close(); + transport.close(); + + const tools = toolsList.tools.map((t: any) => ({ + name: t.name, + description: t.description || 'No description', + inputSchema: t.inputSchema, + })); + + return { + success: true, + serverName, + tools, + count: tools.length, + message: `Found ${tools.length} tool(s) in MCP server '${serverName}'`, + }; + } catch (error) { + return { + success: false, + message: `Failed to list MCP tools: ${error instanceof Error ? error.message : 'Unknown error'}`, + }; + } + }, + }, + + executeCommand: { + description: 'Execute a shell command and return the output. Use this to run bash/shell commands.', + inputSchema: z.object({ + command: z.string().describe('The shell command to execute (e.g., "ls -la", "cat file.txt")'), + cwd: z.string().optional().describe('Working directory to execute the command in (defaults to .rowboat directory)'), + }), + execute: async ({ command, cwd }: { command: string, cwd?: string }) => { + try { + const workingDir = cwd ? path.join(BASE_DIR, cwd) : BASE_DIR; + const result = await executeCommand(command, { cwd: workingDir }); + + return { + success: result.exitCode === 0, + stdout: result.stdout, + stderr: result.stderr, + exitCode: result.exitCode, + command, + workingDir, + }; + } catch (error) { + return { + success: false, + message: `Failed to execute command: ${error instanceof Error ? error.message : 'Unknown error'}`, + command, + }; + } + }, + }, +}; \ No newline at end of file diff --git a/apps/cli/src/application/lib/exec-tool.ts b/apps/cli/src/application/lib/exec-tool.ts index b6485355..d0409365 100644 --- a/apps/cli/src/application/lib/exec-tool.ts +++ b/apps/cli/src/application/lib/exec-tool.ts @@ -1,20 +1,16 @@ -import { tool, Tool } from "ai"; -import { AgentTool } from "../entities/agent.js"; +import { ToolAttachment } from "../entities/agent.js"; import { z } from "zod"; import { McpServers } from "../config/config.js"; -import { getMcpClient } from "./mcp.js"; import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js"; import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; import { SSEClientTransport } from "@modelcontextprotocol/sdk/client/sse.js"; import { Transport } from "@modelcontextprotocol/sdk/shared/transport.js"; import { Client } from "@modelcontextprotocol/sdk/client"; -import { executeCommand } from "./command-executor.js"; -import { loadWorkflow } from "./utils.js"; import { AssistantMessage } from "../entities/message.js"; -import { executeWorkflow } from "./exec-workflow.js"; -import readline from "readline"; +import { BuiltinTools } from "./builtin-tools.js"; +import { streamAgent } from "./agent.js"; -async function execMcpTool(agentTool: z.infer & { type: "mcp" }, input: any): Promise { +async function execMcpTool(agentTool: z.infer & { type: "mcp" }, input: any): Promise { // load mcp configuration from the tool const mcpConfig = McpServers[agentTool.mcpServerName]; if (!mcpConfig) { @@ -57,34 +53,12 @@ async function execMcpTool(agentTool: z.infer & { type: "mcp" return result; } -async function execBashTool(agentTool: z.infer, input: any): Promise { - const result = await executeCommand(input.command as string); - return { - stdout: result.stdout, - stderr: result.stderr, - exitCode: result.exitCode, - }; -} - -export async function execAskHumanTool(agentTool: z.infer, question: string): Promise { - const rl = readline.createInterface({ - input: process.stdin, - output: process.stdout - }); - - let p = new Promise((resolve, reject) => { - rl.question(`>> Provide answer to: ${question}:\n\n`, (answer) => { - resolve(answer); - rl.close(); - }); - }); - const answer = await p; - return answer; -} - -async function execWorkflowTool(agentTool: z.infer & { type: "workflow" }, input: any): Promise { +async function execAgentTool(agentTool: z.infer & { type: "agent" }, input: any): Promise { let lastMsg: z.infer | null = null; - for await (const event of executeWorkflow(agentTool.name, input.message)) { + for await (const event of streamAgent({ + agent: agentTool.name, + input: JSON.stringify(input), + })) { if (event.type === "message" && event.message.role === "assistant") { lastMsg = event.message; } @@ -94,7 +68,7 @@ async function execWorkflowTool(agentTool: z.infer & { type: " } if (!lastMsg) { - throw new Error("No message received from workflow"); + throw new Error("No message received from agent"); } if (typeof lastMsg.content === "string") { return lastMsg.content; @@ -107,18 +81,17 @@ async function execWorkflowTool(agentTool: z.infer & { type: " }, ""); } -export async function execTool(agentTool: z.infer, input: any): Promise { +export async function execTool(agentTool: z.infer, input: any): Promise { switch (agentTool.type) { case "mcp": return execMcpTool(agentTool, input); - case "workflow": - return execWorkflowTool(agentTool, input); + case "agent": + return execAgentTool(agentTool, input); case "builtin": - switch (agentTool.name) { - case "bash": - return execBashTool(agentTool, input); - default: - throw new Error(`Unknown builtin tool: ${agentTool.name}`); + const builtinTool = BuiltinTools[agentTool.name]; + if (!builtinTool || !builtinTool.execute) { + throw new Error(`Unsupported builtin tool: ${agentTool.name}`); } + return builtinTool.execute(input); } } \ No newline at end of file diff --git a/apps/cli/src/application/lib/exec-workflow.ts b/apps/cli/src/application/lib/exec-workflow.ts deleted file mode 100644 index 37421665..00000000 --- a/apps/cli/src/application/lib/exec-workflow.ts +++ /dev/null @@ -1,449 +0,0 @@ -import { loadWorkflow } from "./utils.js"; -import { MessageList, AssistantMessage, AssistantContentPart, Message, ToolMessage, ToolCallPart } from "../entities/message.js"; -import { LlmStepStreamEvent } from "../entities/llm-step-event.js"; -import { AgentNode } from "./agent.js"; -import { z } from "zod"; -import path from "path"; -import { WorkDir } from "../config/config.js"; -import fs from "fs"; -import { createInterface, Interface } from "node:readline/promises"; -import { FunctionsRegistry } from "../registry/functions.js"; -import { RunEvent } from "../entities/workflow-event.js"; -import { execAskHumanTool, execTool } from "./exec-tool.js"; -import { AgentTool } from "../entities/agent.js"; -import { runIdGenerator } from "./run-id-gen.js"; -import { Workflow } from "../entities/workflow.js"; - -const MappedToolCall = z.object({ - toolCall: ToolCallPart, - agentTool: AgentTool, -}); - -const State = z.object({ - stepIndex: z.number(), - messages: MessageList, - workflow: Workflow.nullable(), - pendingToolCallId: z.string().nullable(), -}); - -class StateBuilder { - private state: z.infer = { - stepIndex: 0, - messages: [], - workflow: null, - pendingToolCallId: null, - }; - - ingest(event: z.infer) { - switch (event.type) { - case "start": - this.state.workflow = event.workflow; - break; - case "step-start": - this.state.stepIndex = event.stepIndex; - break; - case "message": - this.state.messages.push(event.message); - this.state.pendingToolCallId = null; - break; - case "pause-for-human-input": - this.state.pendingToolCallId = event.toolCallId; - break; - } - } - - get(): z.infer { - return this.state; - } -} - -class RunLogger { - private logFile: string; - private fileHandle: fs.WriteStream; - - ensureRunsDir(workflowId: string) { - const runsDir = path.join(WorkDir, "runs", workflowId); - if (!fs.existsSync(runsDir)) { - fs.mkdirSync(runsDir, { recursive: true }); - } - } - - constructor(workflowId: string, runId: string) { - this.ensureRunsDir(workflowId); - this.logFile = path.join(WorkDir, "runs", `${runId}.jsonl`); - this.fileHandle = fs.createWriteStream(this.logFile, { - flags: "a", - encoding: "utf8", - }); - } - - log(event: z.infer) { - this.fileHandle.write(JSON.stringify(event) + "\n"); - } - - close() { - this.fileHandle.close(); - } -} - -class LogAndYield { - private logger: RunLogger - - constructor(logger: RunLogger) { - this.logger = logger; - } - - async *logAndYield(event: z.infer): AsyncGenerator, void, unknown> { - const ev = { - ...event, - ts: new Date().toISOString(), - } - this.logger.log(ev); - yield ev; - } -} - -class StreamStepMessageBuilder { - private parts: z.infer[] = []; - private textBuffer: string = ""; - private reasoningBuffer: string = ""; - - flushBuffers() { - if (this.reasoningBuffer) { - this.parts.push({ type: "reasoning", text: this.reasoningBuffer }); - this.reasoningBuffer = ""; - } - if (this.textBuffer) { - this.parts.push({ type: "text", text: this.textBuffer }); - this.textBuffer = ""; - } - } - - ingest(event: z.infer) { - switch (event.type) { - case "reasoning-start": - case "reasoning-end": - case "text-start": - case "text-end": - this.flushBuffers(); - break; - case "reasoning-delta": - this.reasoningBuffer += event.delta; - break; - case "text-delta": - this.textBuffer += event.delta; - break; - case "tool-call": - this.parts.push({ - type: "tool-call", - toolCallId: event.toolCallId, - toolName: event.toolName, - arguments: event.input, - }); - break; - } - } - - get(): z.infer { - this.flushBuffers(); - return { - role: "assistant", - content: this.parts, - }; - } -} - -function loadFunction(id: string) { - const func = FunctionsRegistry[id]; - if (!func) { - throw new Error(`Function ${id} not found`); - } - return func; -} - -export async function* executeWorkflow(id: string, input: string, interactive: boolean = true, asTool: boolean = false): AsyncGenerator, void, unknown> { - const runId = runIdGenerator.next(); - yield* runFromState({ - id, - runId, - state: { - stepIndex: 0, - messages: [{ - role: "user", - content: input, - }], - workflow: null, - pendingToolCallId: null, - }, - interactive, - asTool, - }); -} - -export async function* resumeWorkflow(runId: string, input: string, interactive: boolean = false): AsyncGenerator, void, unknown> { - // read a run.jsonl file line by line and build state - const builder = new StateBuilder(); - let rl: Interface | null = null; - let stream: fs.ReadStream | null = null; - try { - const logFile = path.join(WorkDir, "runs", `${runId}.jsonl`); - stream = fs.createReadStream(logFile, { encoding: "utf8" }); - rl = createInterface({ input: stream, crlfDelay: Infinity }); - for await (const line of rl) { - if (line.trim() === "") { - continue; - } - // console.error('processing line', line); - const parsed = JSON.parse(line); - // console.error('parsed'); - const event = RunEvent.parse(parsed); - // console.error('zod parsed'); - builder.ingest(event); - } - } catch (error) { - // console.error("Failed to resume workflow:", error); - // yield { - // type: "error", - // error: error instanceof Error ? error.message : String(error), - // }; - } finally { - rl?.close(); - stream?.close(); - } - - const { workflow, messages, stepIndex, pendingToolCallId } = builder.get(); - if (!workflow) { - throw new Error(`Workflow not found for run ${runId}`); - } - if (!pendingToolCallId) { - throw new Error(`No pending tool call found for run ${runId}`); - } - const stepId = workflow.steps[stepIndex].id; - - // append user input as message - const logger = new RunLogger(workflow.name, runId); - const ly = new LogAndYield(logger); - yield *ly.logAndYield({ - type: "resume" - }); - - // append user input as message - const resultMsg: z.infer = { - role: "tool", - content: JSON.stringify(input), - toolCallId: pendingToolCallId, - toolName: "ask-human", - }; - messages.push(resultMsg); - yield* ly.logAndYield({ - type: "tool-result", - stepId, - toolName: "ask-human", - result: input, - }); - yield* ly.logAndYield({ - type: "message", - stepId, - message: resultMsg, - }); - - yield* runFromState({ - id: workflow.name, - runId, - state: { - stepIndex, - messages, - workflow, - pendingToolCallId, - }, - interactive, - asTool: false, - }); -} - -async function* runFromState(opts: { - id: string; - runId: string; - state: z.infer; - interactive: boolean; - asTool: boolean; -}) { - const { id, runId, state, interactive, asTool } = opts; - let stepIndex = state.stepIndex; - let messages = [...state.messages]; - let workflow = state.workflow; - - const logger = new RunLogger(id, runId); - const ly = new LogAndYield(logger); - - try { - if (!workflow) { - workflow = loadWorkflow(id); - - yield* ly.logAndYield({ - type: "start", - runId, - workflowId: id, - workflow, - interactive, - }); - } - - while (true) { - const step = workflow.steps[stepIndex]; - const node = step.type === "agent" ? new AgentNode(step.id, asTool) : loadFunction(step.id); - - yield* ly.logAndYield({ - type: "step-start", - stepIndex, - stepId: step.id, - stepType: step.type, - }); - - const messageBuilder = new StreamStepMessageBuilder(); - - // stream response from agent - for await (const event of node.execute(messages)) { - // console.log(" - event", JSON.stringify(event)); - messageBuilder.ingest(event); - yield* ly.logAndYield({ - type: "stream-event", - stepId: step.id, - event: event, - }); - } - - // build and emit final message from agent response - const msg = messageBuilder.get(); - messages.push(msg); - yield* ly.logAndYield({ - type: "message", - stepId: step.id, - message: msg, - }); - - // handle tool calls - const tools = node.tools(); - const mappedToolCalls: z.infer[] = []; - let msgToolCallParts: z.infer[] = []; - if (msg.content instanceof Array) { - msgToolCallParts = msg.content.filter(part => part.type === "tool-call"); - } - const hasToolCalls = msgToolCallParts.length > 0; - - // validate and map tool calls - for (const part of msgToolCallParts) { - const agentTool = tools[part.toolName]; - if (!agentTool) { - throw new Error(`Tool ${part.toolName} not found`); - } - mappedToolCalls.push({ - toolCall: part, - agentTool: agentTool, - }); - } - - // first, exec all tool calls other than ask-human - for (const call of mappedToolCalls) { - const { agentTool, toolCall } = call; - if (agentTool.type === "builtin" && agentTool.name === "ask-human") { - continue; - } - yield* ly.logAndYield({ - type: "tool-invocation", - stepId: step.id, - toolName: toolCall.toolName, - input: JSON.stringify(toolCall.arguments), - }); - const result = await execTool(agentTool, toolCall.arguments); - const resultMsg: z.infer = { - role: "tool", - content: JSON.stringify(result), - toolCallId: toolCall.toolCallId, - toolName: toolCall.toolName, - }; - messages.push(resultMsg); - yield* ly.logAndYield({ - type: "tool-result", - stepId: step.id, - toolName: toolCall.toolName, - result: result, - }); - yield* ly.logAndYield({ - type: "message", - stepId: step.id, - message: resultMsg, - }); - } - - // handle ask-tool call execution - for (const call of mappedToolCalls) { - const { agentTool, toolCall } = call; - if (agentTool.type !== "builtin" || agentTool.name !== "ask-human") { - continue; - } - yield* ly.logAndYield({ - type: "tool-invocation", - stepId: step.id, - toolName: toolCall.toolName, - input: JSON.stringify(toolCall.arguments), - }); - - // if running in background mode, exit here - if (!interactive) { - yield* ly.logAndYield({ - type: "pause-for-human-input", - toolCallId: toolCall.toolCallId, - }); - return; - } - const result = await execAskHumanTool(agentTool, toolCall.arguments.question as string); - const resultMsg: z.infer = { - role: "tool", - content: JSON.stringify(result), - toolCallId: toolCall.toolCallId, - toolName: toolCall.toolName, - }; - messages.push(resultMsg); - yield* ly.logAndYield({ - type: "tool-result", - stepId: step.id, - toolName: toolCall.toolName, - result: result, - }); - yield* ly.logAndYield({ - type: "message", - stepId: step.id, - message: resultMsg, - }); - } - - yield* ly.logAndYield({ - type: "step-end", - stepIndex, - }); - - // if the agent response had tool calls, replay this agent - if (hasToolCalls) { - continue; - } - - // otherwise, move to the next step - stepIndex++; - if (stepIndex >= workflow.steps.length) { - yield* ly.logAndYield({ - type: "end", - }); - break; - } - } - // console.log('\n\n', JSON.stringify(messages, null, 2)); - } catch (error) { - yield* ly.logAndYield({ - type: "error", - error: error instanceof Error ? error.message : String(error), - }); - } finally { - logger.close(); - } -} diff --git a/apps/cli/src/application/lib/step.ts b/apps/cli/src/application/lib/step.ts index ec3c2146..3fae98bc 100644 --- a/apps/cli/src/application/lib/step.ts +++ b/apps/cli/src/application/lib/step.ts @@ -1,7 +1,7 @@ import { MessageList } from "../entities/message.js"; -import { LlmStepStreamEvent } from "../entities/llm-step-event.js"; +import { LlmStepStreamEvent } from "../entities/llm-step-events.js"; import { z } from "zod"; -import { AgentTool } from "../entities/agent.js"; +import { ToolAttachment } from "../entities/agent.js"; export type StepInputT = z.infer; export type StepOutputT = AsyncGenerator, void, unknown>; @@ -9,5 +9,5 @@ export type StepOutputT = AsyncGenerator, voi export interface Step { execute(input: StepInputT): StepOutputT; - tools(): Record>; + tools(): Record>; } \ No newline at end of file diff --git a/apps/cli/src/application/lib/stream-renderer.ts b/apps/cli/src/application/lib/stream-renderer.ts index b5a2cc57..7ac3279d 100644 --- a/apps/cli/src/application/lib/stream-renderer.ts +++ b/apps/cli/src/application/lib/stream-renderer.ts @@ -1,6 +1,6 @@ import { z } from "zod"; -import { RunEvent } from "../entities/workflow-event.js"; -import { LlmStepStreamEvent } from "../entities/llm-step-event.js"; +import { RunEvent } from "../entities/run-events.js"; +import { LlmStepStreamEvent } from "../entities/llm-step-events.js"; export interface StreamRendererOptions { showHeaders?: boolean; @@ -27,11 +27,11 @@ export class StreamRenderer { render(event: z.infer) { switch (event.type) { case "start": { - this.onWorkflowStart(event.workflowId, event.runId, event.interactive); + this.onStart(event.agentId, event.runId, event.interactive); break; } case "step-start": { - this.onStepStart(event.stepIndex, event.stepId, event.stepType); + this.onStepStart(); break; } case "stream-event": { @@ -43,23 +43,23 @@ export class StreamRenderer { break; } case "tool-invocation": { - this.onStepToolInvocation(event.stepId, event.toolName, event.input); + this.onStepToolInvocation(event.toolName, event.input); break; } case "tool-result": { - this.onStepToolResult(event.stepId, event.toolName, event.result); + this.onStepToolResult(event.toolName, event.result); break; } case "step-end": { - this.onStepEnd(event.stepIndex); + this.onStepEnd(); break; } case "end": { - this.onWorkflowEnd(); + this.onEnd(); break; } case "error": { - this.onWorkflowError(event.error); + this.onError(event.error); break; } } @@ -94,29 +94,29 @@ export class StreamRenderer { } } - private onWorkflowStart(workflowId: string, runId: string, interactive: boolean) { + private onStart(workflowId: string, runId: string, interactive: boolean) { this.write("\n"); this.write(this.bold(`ā–¶ Workflow ${workflowId} (run ${runId})`)); if (!interactive) this.write(this.dim(" (--no-interactive)")); this.write("\n"); } - private onWorkflowEnd() { + private onEnd() { this.write(this.bold("\nā–  Workflow complete\n")); } - private onWorkflowError(error: string) { + private onError(error: string) { this.write(this.red(`\nāœ– Workflow error: ${error}\n`)); } - private onStepStart(stepIndex: number, stepId: string, stepType: "agent" | "function") { + private onStepStart() { this.write("\n"); - this.write(this.cyan(`ā”€ Step ${stepIndex} [${stepType}]`)); + this.write(this.cyan(`ā”€ Step started`)); this.write("\n"); } - private onStepEnd(stepIndex: number) { - this.write(this.dim(`āœ“ Step ${stepIndex} finished\n`)); + private onStepEnd() { + this.write(this.dim(`āœ“ Step finished\n`)); } private onStepMessage(stepIndex: number, message: any) { @@ -131,7 +131,7 @@ export class StreamRenderer { } } - private onStepToolInvocation(stepId: string, toolName: string, input: string) { + private onStepToolInvocation(toolName: string, input: string) { this.write(this.cyan(`\nā†’ Tool invoke ${toolName}`)); if (input && input.length) { this.write("\n" + this.dim(this.indent(this.truncate(input))) + "\n"); @@ -140,7 +140,7 @@ export class StreamRenderer { } } - private onStepToolResult(stepId: string, toolName: string, result: unknown) { + private onStepToolResult(toolName: string, result: unknown) { const res = this.truncate(JSON.stringify(result, null, this.options.jsonIndent)); this.write(this.cyan(`\nā† Tool result ${toolName}\n`)); this.write(this.dim(this.indent(res)) + "\n"); diff --git a/apps/cli/src/application/lib/utils.ts b/apps/cli/src/application/lib/utils.ts deleted file mode 100644 index da211b36..00000000 --- a/apps/cli/src/application/lib/utils.ts +++ /dev/null @@ -1,10 +0,0 @@ -import fs from "fs"; -import path from "path"; -import { WorkDir } from "../config/config.js"; -import { Workflow } from "../entities/workflow.js"; - -export function loadWorkflow(id: string) { - const workflowPath = path.join(WorkDir, "workflows", `${id}.json`); - const workflow = fs.readFileSync(workflowPath, "utf8"); - return Workflow.parse(JSON.parse(workflow)); -} diff --git a/apps/cli/src/application/registry/functions.ts b/apps/cli/src/application/registry/functions.ts deleted file mode 100644 index 1d4c1a9b..00000000 --- a/apps/cli/src/application/registry/functions.ts +++ /dev/null @@ -1,6 +0,0 @@ -import { GetDate } from "../functions/get_date.js"; -import { Step } from "../lib/step.js"; - -export const FunctionsRegistry: Record = { - get_date: new GetDate(), -} as const; \ No newline at end of file diff --git a/apps/cli/src/application/registry/tools.ts b/apps/cli/src/application/registry/tools.ts deleted file mode 100644 index e69de29b..00000000 diff --git a/apps/cli/src/x.ts b/apps/cli/src/x.ts deleted file mode 100644 index 9dbd8edb..00000000 --- a/apps/cli/src/x.ts +++ /dev/null @@ -1,8 +0,0 @@ -import { startCopilot } from "./application/assistant/chat.js"; - -export const start = () => { - startCopilot().catch((err) => { - console.error("Failed to run copilot:", err); - process.exitCode = 1; - }); -}