feat(workflows): let email waits count only the execution's own emails

apps/api/src/jobs/email-processor.ts

@@ -28,6 +28,7 @@ import {
   bodyHasListManagementLink,
   buildEmailHeaders,
   classifyEmail,
+  RECIPIENT_OVERRIDE_HEADER,
   senderOneClickUrl,
   withSourceEmail,
 } from '../services/EmailHeaderService.js';
@@ -242,12 +243,12 @@ export async function processEmailJob(emailId: string): Promise<void> {
         : undefined;
 
     // Check for custom recipient override in headers
-    const recipientEmail = customHeaders?.['X-Plunk-Recipient-Override'] || email.contact.email;
+    const recipientEmail = customHeaders?.[RECIPIENT_OVERRIDE_HEADER] || email.contact.email;
 
     // Remove internal headers before sending
     const publicHeaders = customHeaders ? {...customHeaders} : undefined;
-    if (publicHeaders && 'X-Plunk-Recipient-Override' in publicHeaders) {
-      delete publicHeaders['X-Plunk-Recipient-Override'];
+    if (publicHeaders && RECIPIENT_OVERRIDE_HEADER in publicHeaders) {
+      delete publicHeaders[RECIPIENT_OVERRIDE_HEADER];
     }
 
     // Build the outbound headers: standards-based defaults for the email class

apps/api/src/services/EmailHeaderService.ts

@@ -3,6 +3,13 @@ import {EmailSourceType} from '@plunk/db';
 
 import {API_URI} from '../app/constants.js';
 
+/**
+ * Internal header on an email that a workflow step sends to a custom recipient instead of the
+ * contact. The email is still recorded under the contact; the email processor addresses it to this
+ * header's value and strips the header before sending.
+ */
+export const RECIPIENT_OVERRIDE_HEADER = 'X-Plunk-Recipient-Override';
+
 /**
  * Classification of an outbound email, derived by the caller from the email's
  * source type and template/campaign type. This is the single axis that decides

apps/api/src/services/EmailService.ts

@@ -9,7 +9,7 @@ import {ConflictError, ErrorCode, HttpException, NotFound} from '../exceptions/i
 import {createTranslatorSync, renderTemplate} from '@plunk/shared';
 
 import {BillingLimitService} from './BillingLimitService.js';
-import {withSourceEmail} from './EmailHeaderService.js';
+import {RECIPIENT_OVERRIDE_HEADER, withSourceEmail} from './EmailHeaderService.js';
 import {QueueService} from './QueueService.js';
 import {ThrottleService} from './ThrottleService.js';
 
@@ -332,7 +332,7 @@ export class EmailService {
     // If custom recipient email is provided, store it in headers for later use
     const emailHeaders = params.headers ? {...params.headers} : {};
     if (params.recipientEmail) {
-      emailHeaders['X-Plunk-Recipient-Override'] = params.recipientEmail;
+      emailHeaders[RECIPIENT_OVERRIDE_HEADER] = params.recipientEmail;
     }
 
     const email = await prisma.email.create({

apps/api/src/services/EventService.ts

@@ -42,7 +42,7 @@ export class EventService {
     await this.triggerWorkflows(projectId, eventName, contactId, data);
 
     // Resume workflows waiting for this event
-    await WorkflowExecutionService.handleEvent(projectId, eventName, contactId, data);
+    await WorkflowExecutionService.handleEvent(projectId, eventName, contactId, data, emailId);
 
     return event;
   }

apps/api/src/services/WorkflowExecutionService.ts

@@ -1,5 +1,6 @@
 import type {
   Contact,
+  Email,
   Prisma,
   Template,
   Workflow,
@@ -9,7 +10,7 @@ import type {
 } from '@plunk/db';
 import {StepExecutionStatus, WorkflowExecutionStatus} from '@plunk/db';
 import {toPrismaJson} from '@plunk/types';
-import {renderTemplate, WorkflowStepConfigSchemas} from '@plunk/shared';
+import {canScopeWaitToExecution, renderTemplate, WorkflowStepConfigSchemas} from '@plunk/shared';
 import dns from 'node:dns/promises';
 import net from 'node:net';
 import signale from 'signale';
@@ -19,6 +20,7 @@ import {HttpException} from '../exceptions/index.js';
 import {DASHBOARD_URI} from '../app/constants.js';
 
 import {ContactFieldService} from './ContactFieldService.js';
+import {RECIPIENT_OVERRIDE_HEADER} from './EmailHeaderService.js';
 import {EmailService} from './EmailService.js';
 import {NtfyService} from './NtfyService.js';
 import {QueueService} from './QueueService.js';
@@ -471,6 +473,7 @@ export class WorkflowExecutionService {
     eventName: string,
     contactId?: string,
     data?: Record<string, unknown>,
+    emailId?: string,
   ): Promise<void> {
     // Find workflows waiting for this event (a cancelled execution keeps its wait step WAITING)
     const waitingExecutions = await prisma.workflowStepExecution.findMany({
@@ -503,10 +506,29 @@ export class WorkflowExecutionService {
       },
     });
 
+    // The email behind the event, loaded at most once and only for a wait scoped to its execution
+    let sourceEmail: Pick<Email, 'workflowExecutionId' | 'headers'> | null | undefined;
+
     for (const stepExecution of waitingExecutions) {
       const config = stepExecution.step.config;
 
       if (config && typeof config === 'object' && 'eventName' in config && config.eventName === eventName) {
+        // A wait scoped to its execution counts only the emails that execution sent to the contact
+        if ('scope' in config && config.scope === 'execution' && canScopeWaitToExecution(eventName)) {
+          if (sourceEmail === undefined) {
+            sourceEmail = emailId
+              ? await prisma.email.findUnique({
+                  where: {id: emailId},
+                  select: {workflowExecutionId: true, headers: true},
+                })
+              : null;
+          }
+
+          if (!this.sentByExecution(sourceEmail, stepExecution.executionId)) {
+            continue;
+          }
+        }
+
         // Event matches. Claim the wait before resuming it: the list above can be stale by now,
         // as the same event may be delivered twice at once (SES notifications, repeated clicks)
         // or tracked again while an earlier wait of this loop was resumed. Whoever moves the
@@ -543,6 +565,24 @@ export class WorkflowExecutionService {
     }
   }
 
+  /**
+   * Whether an email was sent to the contact by the given execution. An email a step sent to a
+   * custom recipient is recorded under the contact too, but it is not the contact's engagement.
+   */
+  private static sentByExecution(
+    email: Pick<Email, 'workflowExecutionId' | 'headers'> | null,
+    executionId: string,
+  ): boolean {
+    if (email?.workflowExecutionId !== executionId) {
+      return false;
+    }
+
+    const headers =
+      email.headers && typeof email.headers === 'object' && !Array.isArray(email.headers) ? email.headers : {};
+
+    return !(RECIPIENT_OVERRIDE_HEADER in headers);
+  }
+
   /**
    * Execute a specific step based on its type
    */

apps/api/src/services/__tests__/WorkflowExecutionService.waitScope.test.ts

@@ -0,0 +1,143 @@
+import {beforeEach, describe, expect, it, vi} from 'vitest';
+import {EmailSourceType, EmailStatus, StepExecutionStatus, WorkflowExecutionStatus, WorkflowStepType} from '@plunk/db';
+import {prisma as servicePrisma} from '../../database/prisma';
+import {EventService} from '../EventService';
+import {WorkflowExecutionService} from '../WorkflowExecutionService';
+import {factories, getPrismaClient} from '../../../../../test/helpers';
+
+vi.mock('../QueueService', () => ({
+  QueueService: {
+    queueWorkflowStep: vi.fn(async () => ({id: 'mock-job-id'})),
+    queueEmail: vi.fn(async () => ({id: 'mock-email-job-id'})),
+    queueWorkflowTimeout: vi.fn(async () => ({id: 'mock-timeout-job-id'})),
+    cancelWorkflowTimeout: vi.fn(async () => true),
+  },
+}));
+
+/**
+ * By default a WAIT_FOR_EVENT step resumes on the event from any email of the contact.
+ * `scope: "execution"` limits email events to the emails the waiting execution sent the contact.
+ */
+describe('WorkflowExecutionService - WAIT_FOR_EVENT scope', () => {
+  const prisma = getPrismaClient();
+  let projectId: string;
+  let contactId: string;
+
+  beforeEach(async () => {
+    const {project} = await factories.createUserWithProject();
+    projectId = project.id;
+    contactId = (await factories.createContact({projectId, subscribed: true})).id;
+  });
+
+  /** TRIGGER → SEND_EMAIL → WAIT_FOR_EVENT → EXIT, run until it waits. */
+  async function runUntilWaiting(waitConfig: Record<string, unknown>, sendConfig: Record<string, unknown> = {}) {
+    const template = await factories.createTemplate({projectId});
+    const workflow = await factories.createWorkflow({projectId});
+    const trigger = await prisma.workflowStep.findFirstOrThrow({
+      where: {workflowId: workflow.id, type: WorkflowStepType.TRIGGER},
+    });
+    const send = await factories.createWorkflowStep({
+      workflowId: workflow.id,
+      type: WorkflowStepType.SEND_EMAIL,
+      templateId: template.id,
+      config: {templateId: template.id, ...sendConfig},
+    });
+    const wait = await factories.createWorkflowStep({
+      workflowId: workflow.id,
+      type: WorkflowStepType.WAIT_FOR_EVENT,
+      config: {timeout: 3600, ...waitConfig},
+    });
+    const exit = await factories.createWorkflowStep({workflowId: workflow.id, type: WorkflowStepType.EXIT});
+    await prisma.workflowTransition.createMany({
+      data: [
+        {fromStepId: trigger.id, toStepId: send.id},
+        {fromStepId: send.id, toStepId: wait.id},
+        {fromStepId: wait.id, toStepId: exit.id},
+      ],
+    });
+
+    const execution = await prisma.workflowExecution.create({
+      data: {workflowId: workflow.id, contactId, status: WorkflowExecutionStatus.RUNNING, currentStepId: trigger.id},
+    });
+    await WorkflowExecutionService.processStepExecution(execution.id, trigger.id);
+
+    const sent = await prisma.email.findFirstOrThrow({where: {workflowExecutionId: execution.id}});
+    const waitStatus = async () =>
+      (await prisma.workflowStepExecution.findFirstOrThrow({where: {executionId: execution.id, stepId: wait.id}}))
+        .status;
+    expect(await waitStatus()).toBe(StepExecutionStatus.WAITING);
+
+    return {sent, waitStatus};
+  }
+
+  const newsletter = () =>
+    factories.createEmail({
+      projectId,
+      contactId,
+      sourceType: EmailSourceType.CAMPAIGN,
+      status: EmailStatus.SENT,
+      sentAt: new Date(),
+    });
+
+  /** The call the SES webhook makes for an open (Webhooks.receiveSNSWebhook). */
+  const open = (emailId: string) => EventService.trackEvent(projectId, 'email.open', contactId, emailId, {emailId});
+
+  it('without a scope, resumes on an open of any email of the contact, and looks no email up', async () => {
+    const run = await runUntilWaiting({eventName: 'email.open'});
+    const emailLookup = vi.spyOn(servicePrisma.email, 'findUnique');
+
+    await open((await newsletter()).id);
+
+    expect(await run.waitStatus()).toBe(StepExecutionStatus.COMPLETED);
+    expect(emailLookup).not.toHaveBeenCalled();
+  });
+
+  it("with scope execution, ignores other emails and resumes on the execution's own email", async () => {
+    const run = await runUntilWaiting({eventName: 'email.open', scope: 'execution'});
+    const emailLookup = vi.spyOn(servicePrisma.email, 'findUnique');
+
+    await open((await newsletter()).id);
+    expect(await run.waitStatus()).toBe(StepExecutionStatus.WAITING);
+    expect(emailLookup).toHaveBeenCalledTimes(1);
+
+    await open(run.sent.id);
+    expect(await run.waitStatus()).toBe(StepExecutionStatus.COMPLETED);
+  });
+
+  it('with scope execution, ignores an email the execution sent to a custom recipient', async () => {
+    const run = await runUntilWaiting(
+      {eventName: 'email.open', scope: 'execution'},
+      {recipient: {type: 'CUSTOM', customEmail: 'team@plunk.test'}},
+    );
+
+    await open(run.sent.id);
+
+    expect(await run.waitStatus()).toBe(StepExecutionStatus.WAITING);
+  });
+
+  it('with scope execution, ignores an email event that names no email', async () => {
+    const run = await runUntilWaiting({eventName: 'email.open', scope: 'execution'});
+
+    await EventService.trackEvent(projectId, 'email.open', contactId);
+
+    expect(await run.waitStatus()).toBe(StepExecutionStatus.WAITING);
+  });
+
+  it('with scope execution, resumes only the execution that sent the email', async () => {
+    const first = await runUntilWaiting({eventName: 'email.click', scope: 'execution'});
+    const second = await runUntilWaiting({eventName: 'email.click', scope: 'execution'});
+
+    await EventService.trackEvent(projectId, 'email.click', contactId, second.sent.id, {emailId: second.sent.id});
+
+    expect(await first.waitStatus()).toBe(StepExecutionStatus.WAITING);
+    expect(await second.waitStatus()).toBe(StepExecutionStatus.COMPLETED);
+  });
+
+  it('applies no scope to events an execution cannot have produced', async () => {
+    const run = await runUntilWaiting({eventName: 'order.shipped', scope: 'execution'});
+
+    await EventService.trackEvent(projectId, 'order.shipped', contactId);
+
+    expect(await run.waitStatus()).toBe(StepExecutionStatus.COMPLETED);
+  });
+});

apps/mcp/src/tools/workflow-graph.ts

@@ -112,6 +112,7 @@ const STEP_CONFIG_SCHEMAS: Record<BuildableStepType, z.ZodType> = {
     .object({amount: z.number().positive(), unit: z.enum(DELAY_UNITS)})
     .refine((data) => data.amount <= MAX_DELAY[data.unit], {message: 'Delay cannot exceed 365 days'}),
   WAIT_FOR_EVENT: z.object({
+    scope: z.enum(['contact', 'execution']).optional(),
     eventName: z.string().min(1),
     timeout: z.number().nonnegative().max(MAX_WAIT_SECONDS, 'Timeout cannot exceed 365 days').optional(),
   }),

apps/web/src/components/workflow-steps/WaitForEventStepDialog.tsx

@@ -1,4 +1,18 @@
-import {Command, CommandGroup, CommandItem, CommandList, Input, Label, Select, SelectContent, SelectItem, SelectTrigger, SelectValue} from '@plunk/ui';
+import {canScopeWaitToExecution} from '@plunk/shared';
+import {
+  Command,
+  CommandGroup,
+  CommandItem,
+  CommandList,
+  Input,
+  Label,
+  Select,
+  SelectContent,
+  SelectItem,
+  SelectTrigger,
+  SelectValue,
+  Switch,
+} from '@plunk/ui';
 import {useState} from 'react';
 import {toast} from 'sonner';
 import useSWR from 'swr';
@@ -49,6 +63,7 @@ export function WaitForEventStepDialog({step, workflowId, open, onOpenChange, on
   const [eventPopoverOpen, setEventPopoverOpen] = useState(false);
   const [timeoutAmount, setTimeoutAmount] = useState(initialTimeout.amount);
   const [timeoutUnit, setTimeoutUnit] = useState<TimeUnit>(initialUnit);
+  const [onlyThisExecution, setOnlyThisExecution] = useState(config.scope === 'execution');
 
   const {data: eventNamesData} = useSWR<{eventNames: string[]}>(open ? '/events/names' : null, {
     revalidateOnFocus: false,
@@ -72,9 +87,18 @@ export function WaitForEventStepDialog({step, workflowId, open, onOpenChange, on
 
     const timeoutSeconds = amount > 0 ? amount * SECONDS_PER_UNIT[timeoutUnit] : 0;
 
+    // The scope only applies to email events an execution's own emails produce
+    const scoped = onlyThisExecution && canScopeWaitToExecution(eventName);
+
+    // A scoped wait whose email is never sent or opened would otherwise wait forever
+    if (scoped && timeoutSeconds === 0) {
+      toast.error('Set a timeout when waiting only for emails from this run');
+      return;
+    }
+
     const ok = await update({
       name,
-      config: {eventName, timeout: timeoutSeconds},
+      config: {eventName, timeout: timeoutSeconds, ...(scoped ? {scope: 'execution'} : {})},
     });
 
     if (ok) {
@@ -156,6 +180,21 @@ export function WaitForEventStepDialog({step, workflowId, open, onOpenChange, on
           </div>
         </div>
 
+        {canScopeWaitToExecution(eventName) && (
+          <div className="flex items-start gap-3 p-4 bg-neutral-50 rounded-lg border border-neutral-200">
+            <Switch id="editEventScope" checked={onlyThisExecution} onCheckedChange={setOnlyThisExecution} />
+            <div className="flex-1">
+              <Label htmlFor="editEventScope" className="font-medium cursor-pointer">
+                Only emails from this run
+              </Label>
+              <p className="text-xs text-neutral-500 mt-1">
+                Count the event only for emails this workflow sent to the contact in this run. Otherwise any email the
+                contact receives counts. Requires a timeout, in case the email is never sent or opened.
+              </p>
+            </div>
+          </div>
+        )}
+
         <div>
           <Label htmlFor="editEventTimeoutAmount">Timeout (optional)</Label>
           <div className="flex gap-2 mt-1.5">

apps/wiki/content/docs/concepts/workflows.mdx

@@ -66,6 +66,8 @@ A workflow always begins with a single auto-created `TRIGGER` step. You build th
 | `UPDATE_CONTACT`  | Patches contact data — useful for tagging contacts as they progress (`{ stage: "activated" }`).        | `data` object                                                    |
 | `EXIT`            | Terminates the execution. Optionally records an `exitReason` for analytics.                            | optional `reason`                                                |
 
+A `WAIT_FOR_EVENT` step on an email event resumes when that event comes from any email of the contact, including emails sent by campaigns or other workflows (`scope: "contact"`, the default). For the events a workflow email produces (`email.sent`, `email.delivery`, `email.open`, `email.click`, `email.bounce`, `email.complaint`), set `scope: "execution"` in the step config (**Only emails from this run** in the editor) to count only the emails this execution sent to the contact. Emails a step sends to a custom recipient are recorded under the contact but never count for a scoped wait. If the execution sent several emails, an event from any of them resumes the wait. The editor requires a timeout for a scoped wait: if its email is never sent (for example, to an unsubscribed contact) or never opened, the execution would otherwise wait forever.
+
 ## Transitions
 
 Transitions are the edges of the workflow graph. For most step types a transition is a simple "next" pointer. For `CONDITION` steps, each transition carries a `branch` discriminator (`"yes"` or `"no"`) so the engine knows which path to take when the condition evaluates.