Add an option for choosing the debounce mode, trailing or leading, with default leading

Author: ericallam

docs/triggering.mdx

@@ -861,14 +861,37 @@ The `debounce` option accepts:
 
 - `key` - A unique string to identify the debounce group (scoped to the task)
 - `delay` - Duration string specifying how long to delay (e.g., "5s", "1m", "30s")
+- `mode` - Optional. Controls which trigger's data is used: `"leading"` (default) or `"trailing"`
 
 **How it works:**
 
 1. First trigger with a debounce key creates a new delayed run
 2. Subsequent triggers with the same key (while the run is still delayed) push the execution time further
-3. Once no new triggers occur within the delay duration, the run executes with the **first** payload
+3. Once no new triggers occur within the delay duration, the run executes
 4. After the run starts executing, a new trigger with the same key will create a new run
 
+**Leading vs Trailing mode:**
+
+By default, debounce uses **leading mode** - the run executes with data from the **first** trigger.
+
+With **trailing mode**, each subsequent trigger updates the run's data (payload, metadata, tags, maxAttempts, maxDuration, and machine), so the run executes with data from the **last** trigger:
+
+```ts
+// Leading mode (default): runs with first payload
+await myTask.trigger({ count: 1 }, { debounce: { key: "user-123", delay: "5s" } });
+await myTask.trigger({ count: 2 }, { debounce: { key: "user-123", delay: "5s" } });
+// After 5 seconds, runs with { count: 1 }
+
+// Trailing mode: runs with last payload
+await myTask.trigger({ count: 1 }, { debounce: { key: "user-123", delay: "5s", mode: "trailing" } });
+await myTask.trigger({ count: 2 }, { debounce: { key: "user-123", delay: "5s", mode: "trailing" } });
+// After 5 seconds, runs with { count: 2 }
+```
+
+Use **trailing mode** when you want to process the most recent data, such as:
+- Saving the latest version of a document after edits stop
+- Processing the final state after a series of rapid updates
+
 **With `triggerAndWait`:**
 
 When using `triggerAndWait` with debounce, the parent run blocks on the existing debounced run if one exists:

internal-packages/run-engine/src/engine/index.ts

@@ -457,7 +457,22 @@ export class RunEngine {
           const debounceResult = await this.debounceSystem.handleDebounce({
             environmentId: environment.id,
             taskIdentifier,
-            debounce,
+            debounce:
+              debounce.mode === "trailing"
+                ? {
+                    ...debounce,
+                    updateData: {
+                      payload,
+                      payloadType,
+                      metadata,
+                      metadataType,
+                      tags,
+                      maxAttempts,
+                      maxDurationInSeconds,
+                      machine,
+                    },
+                  }
+                : debounce,
             tx: prisma,
           });
 

internal-packages/run-engine/src/engine/systems/debounceSystem.ts

@@ -16,6 +16,18 @@ import { DelayedRunSystem } from "./delayedRunSystem.js";
 export type DebounceOptions = {
   key: string;
   delay: string;
+  mode?: "leading" | "trailing";
+  /** When mode: "trailing", these fields will be used to update the existing run */
+  updateData?: {
+    payload: string;
+    payloadType: string;
+    metadata?: string;
+    metadataType?: string;
+    tags?: { id: string; name: string }[];
+    maxAttempts?: number;
+    maxDurationInSeconds?: number;
+    machine?: string;
+  };
 };
 
 export type DebounceSystemOptions = {
@@ -439,9 +451,24 @@ return { 0, value }
         );
       }
 
+      // Update run data when mode is "trailing"
+      let updatedRun = existingRun;
+      if (debounce.mode === "trailing" && debounce.updateData) {
+        updatedRun = await this.#updateRunForTrailingMode({
+          runId: existingRunId,
+          updateData: debounce.updateData,
+          tx: prisma,
+        });
+
+        this.$.logger.debug("handleExistingRun: updated run data for trailing mode", {
+          existingRunId,
+          debounceKey: debounce.key,
+        });
+      }
+
       return {
         status: "existing",
-        run: existingRun,
+        run: updatedRun,
         waitpoint: existingRun.associatedWaitpoint,
       };
     });
@@ -639,6 +666,75 @@ return { 0, value }
     });
   }
 
+  /**
+   * Updates a run's data for trailing mode debounce.
+   * Updates: payload, metadata, tags, maxAttempts, maxDurationInSeconds, machinePreset
+   */
+  async #updateRunForTrailingMode({
+    runId,
+    updateData,
+    tx,
+  }: {
+    runId: string;
+    updateData: NonNullable<DebounceOptions["updateData"]>;
+    tx?: PrismaClientOrTransaction;
+  }): Promise<TaskRun & { associatedWaitpoint: Waitpoint | null }> {
+    const prisma = tx ?? this.$.prisma;
+
+    // Build the update object
+    const updatePayload: {
+      payload: string;
+      payloadType: string;
+      metadata?: string;
+      metadataType?: string;
+      maxAttempts?: number;
+      maxDurationInSeconds?: number;
+      machinePreset?: string;
+      runTags?: string[];
+      tags?: {
+        set: { id: string }[];
+      };
+    } = {
+      payload: updateData.payload,
+      payloadType: updateData.payloadType,
+    };
+
+    if (updateData.metadata !== undefined) {
+      updatePayload.metadata = updateData.metadata;
+      updatePayload.metadataType = updateData.metadataType ?? "application/json";
+    }
+
+    if (updateData.maxAttempts !== undefined) {
+      updatePayload.maxAttempts = updateData.maxAttempts;
+    }
+
+    if (updateData.maxDurationInSeconds !== undefined) {
+      updatePayload.maxDurationInSeconds = updateData.maxDurationInSeconds;
+    }
+
+    if (updateData.machine !== undefined) {
+      updatePayload.machinePreset = updateData.machine;
+    }
+
+    // Handle tags update - replace existing tags
+    if (updateData.tags !== undefined) {
+      updatePayload.runTags = updateData.tags.map((t) => t.name);
+      updatePayload.tags = {
+        set: updateData.tags.map((t) => ({ id: t.id })),
+      };
+    }
+
+    const updatedRun = await prisma.taskRun.update({
+      where: { id: runId },
+      data: updatePayload,
+      include: {
+        associatedWaitpoint: true,
+      },
+    });
+
+    return updatedRun;
+  }
+
   async quit(): Promise<void> {
     await this.redis.quit();
   }