Refactor pilot server to utilize agent-based architecture for workflow execution and tool management. Update event handling, improve error handling, and enhance workflow-related functions. Adjust SQL queries for better performance and consistency. Update README and other documentation for clarity.

Author: vibegui

mcp-studio/README.md

@@ -73,7 +73,7 @@ The `@ref` syntax wires data between steps:
 
 #### Examples
 
-```json
+```jsonc
 // Direct reference - entire value
 { "user": "@fetch_user" }
 

mcp-studio/server/main.ts

@@ -26,9 +26,7 @@ const runtime = withRuntime<Env, typeof StateSchema, Registry>({
       events: [...WORKFLOW_EVENTS] as string[],
       handler: async ({ events }, env) => {
         try {
-          console.log("handling events", events);
           handleWorkflowEvents(events, env as unknown as Env);
-          console.log("events handled");
           return { success: true };
         } catch (error) {
           console.error(`[MAIN] Error handling events: ${error}`);

mcp-studio/server/stdio-tools.ts

@@ -985,7 +985,7 @@ export async function registerStdioTools(server: McpServer): Promise<void> {
       );
 
       const stepResults = await runSQL<Record<string, unknown>>(
-        "SELECT * FROM workflow_step_result WHERE execution_id = ? ORDER BY created_at ASC",
+        "SELECT * FROM workflow_execution_step_result WHERE execution_id = ? ORDER BY started_at_epoch_ms ASC",
         [args.id],
       );
 
@@ -1097,8 +1097,8 @@ export async function registerStdioTools(server: McpServer): Promise<void> {
         "https://assets.decocache.com/decocms/fd07a578-6b1c-40f1-bc05-88a3b981695d/f7fc4ffa81aec04e37ae670c3cd4936643a7b269.png";
 
       await runSQL(
-        `INSERT INTO assistants (id, title, description, avatar, system_prompt, gateway_id, model, created_at, updated_at, created_by, updated_by)
-         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+        `INSERT INTO assistants (id, title, description, avatar, system_prompt, gateway_id, model, instructions, created_at, updated_at, created_by, updated_by)
+         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
         [
           id,
           args.data.title,
@@ -1107,6 +1107,7 @@ export async function registerStdioTools(server: McpServer): Promise<void> {
           args.data.system_prompt || "",
           args.data.gateway_id || "",
           JSON.stringify(args.data.model || { id: "", connectionId: "" }),
+          "", // instructions - default to empty string
           now,
           now,
           "stdio-user",

mcp-studio/server/tools/workflow.ts

@@ -131,28 +131,32 @@ export const createListTool = (env: Env) =>
         `;
 
       const itemsResult: any =
-        await env.MESH_REQUEST_CONTEXT?.state?.DATABASE.DATABASES_RUN_SQL({
+        await env.MESH_REQUEST_CONTEXT?.state?.DATABASE?.DATABASES_RUN_SQL({
           sql,
           params: [...params, limit, offset],
         });
 
+      if (!itemsResult?.result?.[0]?.results) {
+        throw new Error("Database query failed or returned invalid result");
+      }
+
       const countQuery = `SELECT COUNT(*) as count FROM workflow_collection ${whereClause}`;
       const countResult =
-        await env.MESH_REQUEST_CONTEXT?.state?.DATABASE.DATABASES_RUN_SQL({
+        await env.MESH_REQUEST_CONTEXT?.state?.DATABASE?.DATABASES_RUN_SQL({
           sql: countQuery,
           params,
         });
       const dbTotalCount = parseInt(
-        (countResult.result[0]?.results?.[0] as { count: string })?.count ||
+        (countResult?.result?.[0]?.results?.[0] as { count: string })?.count ||
           "0",
         10,
       );
 
       // Get DB workflows
-      const dbWorkflows: WorkflowWithMeta[] =
-        itemsResult.result[0]?.results?.map((item: Record<string, unknown>) =>
+      const dbWorkflows: WorkflowWithMeta[] = itemsResult.result[0].results.map(
+        (item: Record<string, unknown>) =>
           transformDbRowToWorkflowCollectionItem(item),
-        ) || [];
+      );
 
       // Get file-based workflows (always included, marked readonly)
       const fileWorkflows = getFileWorkflows();
@@ -183,11 +187,11 @@ export async function getWorkflowCollection(
 ): Promise<WorkflowWithMeta | null> {
   // First check DB
   const result =
-    await env.MESH_REQUEST_CONTEXT?.state?.DATABASE.DATABASES_RUN_SQL({
+    await env.MESH_REQUEST_CONTEXT?.state?.DATABASE?.DATABASES_RUN_SQL({
       sql: "SELECT * FROM workflow_collection WHERE id = ? LIMIT 1",
       params: [id],
     });
-  const item = result.result[0]?.results?.[0] || null;
+  const item = result?.result?.[0]?.results?.[0] || null;
 
   if (item) {
     return transformDbRowToWorkflowCollectionItem(
@@ -294,7 +298,7 @@ Example workflow with 2 parallel steps:
 
 Example workflow with a step that references the output of another step:
 { "title": "Get first user and then fetch orders", "steps": [
-  true }, "transformCode": "export default async (i) => i[0]" },
+  { "name": "fetch_users", "action": { "toolName": "GET_USERS" }, "input": { "all": true }, "transformCode": "export default async (i) => i[0]" },
   { "name": "fetch_orders", "action": { "toolName": "GET_ORDERS" }, "input": { "user": "@fetch_users.user" } },
 ]}
 `,
@@ -395,18 +399,18 @@ async function updateWorkflowCollection(
       `;
 
   const result =
-    await env.MESH_REQUEST_CONTEXT?.state?.DATABASE.DATABASES_RUN_SQL({
+    await env.MESH_REQUEST_CONTEXT?.state?.DATABASE?.DATABASES_RUN_SQL({
       sql,
       params,
     });
 
-  if (result.result[0]?.results?.length === 0) {
+  if (!result?.result?.[0]?.results || result.result[0].results.length === 0) {
     throw new Error(`Workflow collection with id ${id} not found`);
   }
 
   return {
     item: transformDbRowToWorkflowCollectionItem(
-      result.result[0]?.results?.[0] as Record<string, unknown>,
+      result.result[0].results[0] as Record<string, unknown>,
     ),
   };
 }

mcp-studio/server/workflow-loader.ts

@@ -83,7 +83,15 @@ export function isFilesystemMode(): boolean {
  * Parse a workflow file and extract workflow(s)
  */
 async function parseWorkflowFile(filePath: string): Promise<LoadedWorkflow[]> {
-  const content = await readFile(filePath, "utf-8");
+  let content: string;
+  try {
+    content = await readFile(filePath, "utf-8");
+  } catch (error) {
+    // Handle file access errors (deleted, permission denied, etc.) gracefully
+    console.error(`[workflow-loader] Failed to read ${filePath}:`, error);
+    return [];
+  }
+
   let parsed: unknown;
 
   try {
@@ -254,7 +262,7 @@ export function getWorkflowById(id: string): LoadedWorkflow | undefined {
  * Start watching for file changes
  */
 export async function startWatching(
-  options: WorkflowLoaderOptions,
+  options?: WorkflowLoaderOptions,
 ): Promise<void> {
   const source = options || getWorkflowSource();
 

pilot/README.md

@@ -282,7 +282,7 @@ bun run check
 ## See Also
 
 - [Architecture](docs/ARCHITECTURE.md) - Detailed architecture overview
-- [Mesh Bridge](../../mesh-bridge) - Browser interface for Pilot
+- [Mesh Bridge](../mesh-bridge) - Browser interface for Pilot
 - [MCP Mesh](https://github.com/decolabs/mesh) - The mesh platform
 
 ## License

pilot/server/agent.ts

@@ -116,8 +116,8 @@ export class PilotAgent {
       temperature: 0.7,
       maxRouterIterations: 10,
       maxExecutorIterations: 30,
-      smartModel: config.smartModel || config.fastModel,
       ...config,
+      smartModel: config.smartModel ?? config.fastModel,
     };
     this.ctx = ctx;
     this.localTools = getAllLocalTools();

pilot/server/core/execution-adapter.ts

@@ -190,10 +190,12 @@ export async function findContinuableThread(
     if (exec.status !== "success") continue;
 
     // Must be within TTL
-    if (exec.completed_at_epoch_ms) {
-      const age = now - exec.completed_at_epoch_ms;
-      if (age > ttlMs) continue;
+    if (!exec.completed_at_epoch_ms) {
+      // Missing completion timestamp - skip (incomplete execution)
+      continue;
     }
+    const age = now - exec.completed_at_epoch_ms;
+    if (age > ttlMs) continue;
 
     // Check metadata
     const input = exec.input || {};

pilot/server/core/llm-executor.ts

@@ -12,7 +12,6 @@ import {
   resolveRefs,
   groupStepsByLevel,
 } from "../types/workflow.ts";
-import { loadWorkflow, listWorkflows } from "./workflow-studio-adapter.ts";
 import { getAllLocalTools } from "../tools/index.ts";
 
 // ============================================================================
@@ -52,6 +51,8 @@ export interface ExecutorConfig {
   fastModel: string;
   smartModel: string;
   onProgress?: (stepName: string, message: string) => void;
+  loadWorkflow?: (workflowId: string) => Promise<Workflow | null>;
+  listWorkflows?: () => Promise<Workflow[]>;
 }
 
 export interface ExecutionContext {
@@ -64,6 +65,8 @@ export interface ExecutionContext {
   listConnections: ListConnectionsCallback;
   publishEvent?: (type: string, data: Record<string, unknown>) => Promise<void>;
   toolCache: Map<string, ToolDefinition>;
+  loadWorkflow?: (workflowId: string) => Promise<Workflow | null>;
+  listWorkflows?: () => Promise<Workflow[]>;
 }
 
 export interface WorkflowResult {
@@ -92,9 +95,17 @@ export async function runWorkflow(
       type: string,
       data: Record<string, unknown>,
     ) => Promise<void>;
+    loadWorkflow?: (workflowId: string) => Promise<Workflow | null>;
   },
 ): Promise<WorkflowResult> {
-  const workflow = await loadWorkflow(workflowId);
+  if (!options.loadWorkflow) {
+    return {
+      success: false,
+      error: "loadWorkflow callback not provided",
+    };
+  }
+
+  const workflow = await options.loadWorkflow(workflowId);
   if (!workflow) {
     return { success: false, error: `Workflow not found: ${workflowId}` };
   }
@@ -117,6 +128,8 @@ export async function runWorkflowDirect(
       type: string,
       data: Record<string, unknown>,
     ) => Promise<void>;
+    loadWorkflow?: (workflowId: string) => Promise<Workflow | null>;
+    listWorkflows?: () => Promise<Workflow[]>;
   },
 ): Promise<WorkflowResult> {
   const ctx: ExecutionContext = {
@@ -129,6 +142,8 @@ export async function runWorkflowDirect(
     listConnections: options.listConnections,
     publishEvent: options.publishEvent,
     toolCache: new Map(),
+    loadWorkflow: options.loadWorkflow,
+    listWorkflows: options.listWorkflows,
   };
 
   try {
@@ -266,6 +281,10 @@ async function executeCodeStep(
 
   const code = step.action.code;
 
+  // SECURITY WARNING: Using new Function() is equivalent to eval() and provides
+  // full Node.js access without sandboxing. This should only be used with trusted workflows.
+  // Consider using a sandboxed execution environment (vm2, isolated-vm, or worker threads)
+  // for untrusted code execution in the future.
   try {
     const fn = new Function(
       "input",
@@ -585,7 +604,10 @@ async function executeToolCall(
 ): Promise<unknown> {
   // Meta tools
   if (toolName === "list_workflows") {
-    const workflows = await listWorkflows();
+    if (!ctx.listWorkflows) {
+      throw new Error("listWorkflows callback not provided");
+    }
+    const workflows = await ctx.listWorkflows();
     return {
       workflows: workflows.map((w) => ({
         id: w.id,
@@ -596,6 +618,9 @@ async function executeToolCall(
   }
 
   if (toolName === "start_workflow") {
+    if (!ctx.loadWorkflow) {
+      throw new Error("loadWorkflow callback not provided");
+    }
     const workflowId = args.workflowId as string;
     const workflowInput = (args.input as Record<string, unknown>) || {};
     const result = await runWorkflow(workflowId, workflowInput, {
@@ -604,6 +629,7 @@ async function executeToolCall(
       callMeshTool: ctx.callMeshTool,
       listConnections: ctx.listConnections,
       publishEvent: ctx.publishEvent,
+      loadWorkflow: ctx.loadWorkflow,
     });
     return result;
   }

pilot/server/core/workflow-studio-adapter.ts

@@ -200,6 +200,17 @@ export async function deleteWorkflow(workflowId: string): Promise<boolean> {
   const client = requireStudioClient();
 
   try {
+    // Check if workflow exists and is readonly (file-based)
+    const existingResult = (await client.callTool("COLLECTION_WORKFLOW_GET", {
+      id: workflowId,
+    })) as { item?: { readonly?: boolean } };
+
+    if (existingResult.item?.readonly) {
+      throw new Error(
+        `Cannot delete "${workflowId}" - it's a file-based workflow. Delete the JSON file directly.`,
+      );
+    }
+
     await client.callTool("COLLECTION_WORKFLOW_DELETE", {
       id: workflowId,
     });