feat: Workflow execution snapshotting for safe live editing

apps/api/src/controllers/Workflows.ts

@@ -305,6 +305,26 @@ export class Workflows {
     return res.status(201).json(execution);
   }
 
+  /**
+   * GET /workflows/:id/step-execution-counts
+   * Get active execution counts per step for a workflow
+   */
+  @Get(':id/step-execution-counts')
+  @Middleware([requireAuth, requireEmailVerified])
+  @CatchAsync
+  public async getStepExecutionCounts(req: Request, res: Response, _next: NextFunction) {
+    const auth = res.locals.auth;
+    const workflowId = req.params.id;
+
+    if (!workflowId) {
+      return res.status(400).json({error: 'Workflow ID is required'});
+    }
+
+    const result = await WorkflowService.getStepExecutionCounts(auth.projectId!, workflowId);
+
+    return res.status(200).json(result);
+  }
+
   /**
    * GET /workflows/:id/executions
    * List executions for a workflow

apps/api/src/services/ActivityService.ts

@@ -744,7 +744,7 @@ export class ActivityService {
     });
 
     return stepExecutions
-      .filter(stepExec => stepExec.step.type === 'SEND_EMAIL' && stepExec.scheduledFor)
+      .filter(stepExec => stepExec.step?.type === 'SEND_EMAIL' && stepExec.scheduledFor)
       .map(stepExec => ({
         id: `workflow_step_${stepExec.id}_scheduled`,
         type: ActivityType.WORKFLOW_EMAIL_SCHEDULED,
@@ -753,9 +753,9 @@ export class ActivityService {
         contactId: stepExec.execution.contactId,
         metadata: {
           workflowName: stepExec.execution.workflow.name,
-          stepName: stepExec.step.name,
+          stepName: stepExec.step?.name,
           subject:
-            stepExec.step.config &&
+            stepExec.step?.config &&
             typeof stepExec.step.config === 'object' &&
             stepExec.step.config !== null &&
             'subject' in stepExec.step.config &&

apps/api/src/services/EventService.ts

@@ -1,7 +1,7 @@
 import type {Event} from '@plunk/db';
 import {Prisma} from '@plunk/db';
 import type {FilterCondition, FilterGroup} from '@plunk/types';
-import {toPrismaJson} from '@plunk/types';
+import {buildWorkflowSnapshot, toPrismaJson} from '@plunk/types';
 import signale from 'signale';
 
 import {prisma} from '../database/prisma.js';
@@ -466,6 +466,22 @@ export class EventService {
         return;
       }
 
+      // Build workflow snapshot — freeze the graph so in-flight executions are immune to live edits
+      const fullWorkflow = await prisma.workflow.findUniqueOrThrow({
+        where: {id: workflowId},
+        include: {
+          steps: {
+            include: {
+              template: {
+                select: {id: true, subject: true, body: true, from: true, fromName: true, replyTo: true},
+              },
+              outgoingTransitions: true,
+            },
+          },
+        },
+      });
+      const snapshot = buildWorkflowSnapshot(fullWorkflow);
+
       // Create workflow execution
       const execution = await prisma.workflowExecution.create({
         data: {
@@ -474,6 +490,7 @@ export class EventService {
           status: 'RUNNING',
           currentStepId: triggerStep.id,
           context: context ? toPrismaJson(context) : undefined,
+          workflowSnapshot: toPrismaJson(snapshot),
         },
       });