feat(copilot): copilot mcp + server side copilot execution (#3173)

* v0

* v1

* Basic ss tes

* Ss tests

* Stuff

* Add mcp

* mcp v1

* Improvement

* Fix

* BROKEN

* Checkpoint

* Streaming

* Fix abort

* Things are broken

* Streaming seems to work but copilot is dumb

* Fix edge issue

* LUAAAA

* Fix stream buffer

* Fix lint

* Checkpoint

* Initial temp state, in the middle of a refactor

* Initial test shows diff store still working

* Tool refactor

* First cleanup pass complete - untested

* Continued cleanup

* Refactor

* Refactor complete - no testing yet

* Fix - cursor makes me sad

* Fix mcp

* Clean up mcp

* Updated mcp

* Add respond to subagents

* Fix definitions

* Add tools

* Add tools

* Add copilot mcp tracking

* Fix lint

* Fix mcp

* Fix

* Updates

* Clean up mcp

* Fix copilot mcp tool names to be sim prefixed

* Add opus 4.6

* Fix discovery tool

* Fix

* Remove logs

* Fix go side tool rendering

* Update docs

* Fix hydration

* Fix tool call resolution

* Fix

* Fix lint

* Fix superagent and autoallow integrations

* Fix always allow

* Update block

* Remove plan docs

* Fix hardcoded ff

* Fix dropped provider

* Fix lint

* Fix tests

* Fix dead messages array

* Fix discovery

* Fix run workflow

* Fix run block

* Fix run from block in copilot

* Fix lint

* Fix skip and mtb

* Fix typing

* Fix tool call

* Bump api version

* Fix bun lock

* Nuke bad files

apps/sim/lib/workflows/executor/execute-workflow.ts

@@ -4,7 +4,7 @@ import { LoggingSession } from '@/lib/logs/execution/logging-session'
 import { executeWorkflowCore } from '@/lib/workflows/executor/execution-core'
 import { PauseResumeManager } from '@/lib/workflows/executor/human-in-the-loop-manager'
 import { ExecutionSnapshot } from '@/executor/execution/snapshot'
-import type { ExecutionMetadata } from '@/executor/execution/types'
+import type { ExecutionMetadata, SerializableExecutionState } from '@/executor/execution/types'
 import type { ExecutionResult, StreamingExecution } from '@/executor/types'
 
 const logger = createLogger('WorkflowExecution')
@@ -20,6 +20,15 @@ export interface ExecuteWorkflowOptions {
   includeFileBase64?: boolean
   base64MaxBytes?: number
   abortSignal?: AbortSignal
+  /** Use the live/draft workflow state instead of the deployed state. Used by copilot. */
+  useDraftState?: boolean
+  /** Stop execution after this block completes. Used for "run until block" feature. */
+  stopAfterBlockId?: string
+  /** Run-from-block configuration using a prior execution snapshot. */
+  runFromBlock?: {
+    startBlockId: string
+    sourceSnapshot: SerializableExecutionState
+  }
 }
 
 export interface WorkflowInfo {
@@ -57,7 +66,7 @@ export async function executeWorkflow(
       userId: actorUserId,
       workflowUserId: workflow.userId,
       triggerType,
-      useDraftState: false,
+      useDraftState: streamConfig?.useDraftState ?? false,
       startTime: new Date().toISOString(),
       isClientSession: false,
     }
@@ -84,6 +93,8 @@ export async function executeWorkflow(
       includeFileBase64: streamConfig?.includeFileBase64,
       base64MaxBytes: streamConfig?.base64MaxBytes,
       abortSignal: streamConfig?.abortSignal,
+      stopAfterBlockId: streamConfig?.stopAfterBlockId,
+      runFromBlock: streamConfig?.runFromBlock,
     })
 
     if (result.status === 'paused') {

apps/sim/lib/workflows/executor/execution-core.ts

@@ -400,6 +400,7 @@ export async function executeWorkflowCore(
       finalOutput: result.output || {},
       traceSpans: traceSpans || [],
       workflowInput: processedInput,
+      executionState: result.executionState,
     })
 
     await clearExecutionCancellation(executionId)

apps/sim/lib/workflows/executor/execution-state.ts

@@ -0,0 +1,53 @@
+import { db } from '@sim/db'
+import { workflowExecutionLogs } from '@sim/db/schema'
+import { and, desc, eq, sql } from 'drizzle-orm'
+import type { SerializableExecutionState } from '@/executor/execution/types'
+
+function isSerializableExecutionState(value: unknown): value is SerializableExecutionState {
+  if (!value || typeof value !== 'object') return false
+  const state = value as Record<string, unknown>
+  return (
+    typeof state.blockStates === 'object' &&
+    Array.isArray(state.executedBlocks) &&
+    Array.isArray(state.blockLogs) &&
+    typeof state.decisions === 'object' &&
+    Array.isArray(state.completedLoops) &&
+    Array.isArray(state.activeExecutionPath)
+  )
+}
+
+function extractExecutionState(executionData: unknown): SerializableExecutionState | null {
+  if (!executionData || typeof executionData !== 'object') return null
+  const state = (executionData as Record<string, unknown>).executionState
+  return isSerializableExecutionState(state) ? state : null
+}
+
+export async function getExecutionState(
+  executionId: string
+): Promise<SerializableExecutionState | null> {
+  const [row] = await db
+    .select({ executionData: workflowExecutionLogs.executionData })
+    .from(workflowExecutionLogs)
+    .where(eq(workflowExecutionLogs.executionId, executionId))
+    .limit(1)
+
+  return extractExecutionState(row?.executionData)
+}
+
+export async function getLatestExecutionState(
+  workflowId: string
+): Promise<SerializableExecutionState | null> {
+  const [row] = await db
+    .select({ executionData: workflowExecutionLogs.executionData })
+    .from(workflowExecutionLogs)
+    .where(
+      and(
+        eq(workflowExecutionLogs.workflowId, workflowId),
+        sql`${workflowExecutionLogs.executionData} -> 'executionState' IS NOT NULL`
+      )
+    )
+    .orderBy(desc(workflowExecutionLogs.startedAt))
+    .limit(1)
+
+  return extractExecutionState(row?.executionData)
+}

apps/sim/lib/workflows/utils.ts

@@ -1,7 +1,7 @@
 import { db } from '@sim/db'
 import { permissions, userStats, workflow as workflowTable } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
-import { and, eq } from 'drizzle-orm'
+import { and, asc, eq, inArray, or } from 'drizzle-orm'
 import { NextResponse } from 'next/server'
 import { getSession } from '@/lib/auth'
 import { getWorkspaceWithOwner, type PermissionType } from '@/lib/workspaces/permissions/utils'
@@ -15,6 +15,53 @@ export async function getWorkflowById(id: string) {
   return rows[0]
 }
 
+export async function resolveWorkflowIdForUser(
+  userId: string,
+  workflowId?: string,
+  workflowName?: string
+): Promise<{ workflowId: string; workflowName?: string } | null> {
+  if (workflowId) {
+    return { workflowId }
+  }
+
+  const workspaceIds = await db
+    .select({ entityId: permissions.entityId })
+    .from(permissions)
+    .where(and(eq(permissions.userId, userId), eq(permissions.entityType, 'workspace')))
+
+  const workspaceIdList = workspaceIds.map((row) => row.entityId)
+
+  const workflowConditions = [eq(workflowTable.userId, userId)]
+  if (workspaceIdList.length > 0) {
+    workflowConditions.push(inArray(workflowTable.workspaceId, workspaceIdList))
+  }
+
+  const workflows = await db
+    .select()
+    .from(workflowTable)
+    .where(or(...workflowConditions))
+    .orderBy(asc(workflowTable.sortOrder), asc(workflowTable.createdAt), asc(workflowTable.id))
+
+  if (workflows.length === 0) {
+    return null
+  }
+
+  if (workflowName) {
+    const match = workflows.find(
+      (w) =>
+        String(w.name || '')
+          .trim()
+          .toLowerCase() === workflowName.toLowerCase()
+    )
+    if (match) {
+      return { workflowId: match.id, workflowName: match.name || undefined }
+    }
+    return null
+  }
+
+  return { workflowId: workflows[0].id, workflowName: workflows[0].name || undefined }
+}
+
 type WorkflowRecord = ReturnType<typeof getWorkflowById> extends Promise<infer R>
   ? NonNullable<R>
   : never