feat(webapp): add skipIfActive trigger option for drop-on-conflict dedup

apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts

@@ -135,6 +135,7 @@ const { action, loader } = createActionApiRoute(
         {
           id: result.run.friendlyId,
           isCached: result.isCached,
+          ...(result.wasSkipped ? { wasSkipped: true } : {}),
         },
         {
           headers: $responseHeaders,

apps/webapp/app/runEngine/concerns/skipIfActive.server.ts

@@ -0,0 +1,105 @@
+import { Prisma, type PrismaClientOrTransaction, type TaskRun } from "@trigger.dev/database";
+import { logger } from "~/services/logger.server";
+import type { TriggerTaskRequest } from "../types";
+
+export type SkipIfActiveConcernResult =
+  | { wasSkipped: true; run: TaskRun }
+  | { wasSkipped: false };
+
+/**
+ * DB-level `TaskRunStatus` values that represent a run that has not reached
+ * a terminal state — i.e. still counts as "active" for dedup purposes.
+ * Mirrors the non-final statuses in `TaskRunStatus` (see
+ * `internal-packages/database/prisma/schema.prisma`).
+ */
+const ACTIVE_TASK_RUN_STATUSES = [
+  "DELAYED",
+  "PENDING",
+  "PENDING_VERSION",
+  "WAITING_FOR_DEPLOY",
+  "DEQUEUED",
+  "EXECUTING",
+  "WAITING_TO_RESUME",
+  "RETRYING_AFTER_FAILURE",
+  "PAUSED",
+] as const;
+
+/**
+ * Implements the `skipIfActive` trigger option.
+ *
+ * When `body.options.skipIfActive === true` and at least one tag is set, the
+ * concern looks for any in-flight TaskRun with:
+ *
+ *   runtimeEnvironmentId = <env>
+ *   taskIdentifier       = <task>
+ *   status IN (non-terminal)
+ *   runTags           @> <supplied tags>
+ *
+ * If found, the existing run is returned and the trigger is short-circuited.
+ * If not found, the caller proceeds to create a new run as usual.
+ *
+ * Intended use case: cron-style scanners that poll at a fixed cadence but
+ * should drop duplicate triggers while a prior invocation is still running —
+ * without generating queue backlog (`concurrencyKey`) or caching successful
+ * completions (`idempotencyKey`).
+ */
+export class SkipIfActiveConcern {
+  constructor(private readonly prisma: PrismaClientOrTransaction) {}
+
+  async handleTriggerRequest(request: TriggerTaskRequest): Promise<SkipIfActiveConcernResult> {
+    if (request.body.options?.skipIfActive !== true) {
+      return { wasSkipped: false };
+    }
+
+    const rawTags = request.body.options?.tags;
+    const tags = Array.isArray(rawTags) ? rawTags : rawTags ? [rawTags] : [];
+
+    if (tags.length === 0) {
+      // `skipIfActive` requires a tag scope — without tags, every run of this
+      // task would dedup against every other, which is rarely the intent.
+      // Treat as no-op rather than silently matching.
+      logger.debug("[SkipIfActiveConcern] skipIfActive=true with no tags — skipping the check", {
+        taskIdentifier: request.taskId,
+      });
+      return { wasSkipped: false };
+    }
+
+    // `runTags @> ARRAY[...]::text[]` hits the GIN ArrayOps index on
+    // `TaskRun.runTags` and AND-matches every supplied tag. Bounded by
+    // runtimeEnvironmentId + taskIdentifier + status via existing indexes.
+    const statusArray = Prisma.sql`ARRAY[${Prisma.join(
+      ACTIVE_TASK_RUN_STATUSES.map((s) => Prisma.sql`${s}`)
+    )}]::"TaskRunStatus"[]`;
+
+    const existing = await this.prisma.$queryRaw<Array<{ id: string }>>`
+      SELECT id
+      FROM "TaskRun"
+      WHERE "runtimeEnvironmentId" = ${request.environment.id}
+        AND "taskIdentifier" = ${request.taskId}
+        AND status = ANY(${statusArray})
+        AND "runTags" @> ${tags}::text[]
+      LIMIT 1
+    `;
+
+    if (existing.length === 0) {
+      return { wasSkipped: false };
+    }
+
+    const run = await this.prisma.taskRun.findUnique({ where: { id: existing[0].id } });
+    if (!run) {
+      // Row disappeared between the existence probe and the fetch (e.g.
+      // completed + deleted mid-query). Treat as "no active run" so the
+      // caller creates a fresh one instead of failing.
+      return { wasSkipped: false };
+    }
+
+    logger.debug("[SkipIfActiveConcern] active run matched, skipping new trigger", {
+      runId: run.id,
+      taskIdentifier: request.taskId,
+      environmentId: request.environment.id,
+      tags,
+    });
+
+    return { wasSkipped: true, run };
+  }
+}

apps/webapp/app/runEngine/services/triggerTask.server.ts

@@ -31,6 +31,7 @@ import type {
 } from "../../v3/services/triggerTask.server";
 import { clampMaxDuration } from "../../v3/utils/maxDuration";
 import { IdempotencyKeyConcern } from "../concerns/idempotencyKeys.server";
+import { SkipIfActiveConcern } from "../concerns/skipIfActive.server";
 import type {
   PayloadProcessor,
   QueueManager,
@@ -54,6 +55,7 @@ export class RunEngineTriggerTaskService {
   private readonly validator: TriggerTaskValidator;
   private readonly payloadProcessor: PayloadProcessor;
   private readonly idempotencyKeyConcern: IdempotencyKeyConcern;
+  private readonly skipIfActiveConcern: SkipIfActiveConcern;
   private readonly runNumberIncrementer: RunNumberIncrementer;
   private readonly prisma: PrismaClientOrTransaction;
   private readonly engine: RunEngine;
@@ -69,6 +71,7 @@ export class RunEngineTriggerTaskService {
     validator: TriggerTaskValidator;
     payloadProcessor: PayloadProcessor;
     idempotencyKeyConcern: IdempotencyKeyConcern;
+    skipIfActiveConcern?: SkipIfActiveConcern;
     runNumberIncrementer: RunNumberIncrementer;
     traceEventConcern: TraceEventConcern;
     tracer: Tracer;
@@ -81,6 +84,7 @@ export class RunEngineTriggerTaskService {
     this.validator = opts.validator;
     this.payloadProcessor = opts.payloadProcessor;
     this.idempotencyKeyConcern = opts.idempotencyKeyConcern;
+    this.skipIfActiveConcern = opts.skipIfActiveConcern ?? new SkipIfActiveConcern(opts.prisma);
     this.runNumberIncrementer = opts.runNumberIncrementer;
     this.tracer = opts.tracer;
     this.traceEventConcern = opts.traceEventConcern;
@@ -207,6 +211,14 @@ export class RunEngineTriggerTaskService {
 
       const { idempotencyKey, idempotencyKeyExpiresAt } = idempotencyKeyConcernResult;
 
+      // `skipIfActive` — drop-on-conflict. Runs *after* idempotency so a
+      // deliberate idempotency cache-hit wins, but *before* run creation so
+      // we never touch the queue for a skipped trigger.
+      const skipIfActiveResult = await this.skipIfActiveConcern.handleTriggerRequest(triggerRequest);
+      if (skipIfActiveResult.wasSkipped) {
+        return { run: skipIfActiveResult.run, isCached: true, wasSkipped: true };
+      }
+
       if (idempotencyKey) {
         await this.triggerRacepointSystem.waitForRacepoint({
           racepoint: "idempotencyKey",

apps/webapp/app/v3/services/triggerTask.server.ts

@@ -45,6 +45,12 @@ export class OutOfEntitlementError extends Error {
 export type TriggerTaskServiceResult = {
   run: TaskRun;
   isCached: boolean;
+  /**
+   * True when the run returned was matched by the `skipIfActive` option and
+   * no new run was created. Always false/undefined for idempotency-cached
+   * runs — `isCached` distinguishes those.
+   */
+  wasSkipped?: boolean;
 };
 
 export const MAX_ATTEMPTS = 2;

apps/webapp/test/engine/skipIfActiveConcern.test.ts

@@ -0,0 +1,144 @@
+import { describe, expect, it, vi } from "vitest";
+
+vi.mock("~/services/logger.server", () => ({
+  logger: {
+    debug: vi.fn(),
+    info: vi.fn(),
+    warn: vi.fn(),
+    error: vi.fn(),
+  },
+}));
+
+import { SkipIfActiveConcern } from "../../app/runEngine/concerns/skipIfActive.server";
+import type { TriggerTaskRequest } from "../../app/runEngine/types";
+
+type MockPrisma = {
+  $queryRaw: ReturnType<typeof vi.fn>;
+  taskRun: { findUnique: ReturnType<typeof vi.fn> };
+};
+
+function buildRequest(overrides: {
+  skipIfActive?: boolean;
+  tags?: string | string[];
+  taskId?: string;
+  environmentId?: string;
+}): TriggerTaskRequest {
+  return {
+    taskId: overrides.taskId ?? "ezderm-notes-fetch",
+    friendlyId: "run_test",
+    environment: {
+      id: overrides.environmentId ?? "env_123",
+      organizationId: "org_1",
+      projectId: "proj_1",
+    } as TriggerTaskRequest["environment"],
+    body: {
+      payload: {},
+      options: {
+        skipIfActive: overrides.skipIfActive,
+        tags: overrides.tags,
+      },
+    },
+    options: {},
+  } as TriggerTaskRequest;
+}
+
+function mockPrisma(initial?: {
+  existing?: Array<{ id: string }>;
+  run?: { id: string } | null;
+}): MockPrisma {
+  return {
+    $queryRaw: vi.fn().mockResolvedValue(initial?.existing ?? []),
+    taskRun: { findUnique: vi.fn().mockResolvedValue(initial?.run ?? null) },
+  };
+}
+
+describe("SkipIfActiveConcern", () => {
+  it("returns wasSkipped=false when the flag is not set", async () => {
+    const prisma = mockPrisma();
+    const concern = new SkipIfActiveConcern(prisma as never);
+
+    const result = await concern.handleTriggerRequest(
+      buildRequest({ skipIfActive: undefined, tags: ["connector:abc"] })
+    );
+
+    expect(result).toEqual({ wasSkipped: false });
+    expect(prisma.$queryRaw).not.toHaveBeenCalled();
+  });
+
+  it("returns wasSkipped=false when skipIfActive=false", async () => {
+    const prisma = mockPrisma();
+    const concern = new SkipIfActiveConcern(prisma as never);
+
+    const result = await concern.handleTriggerRequest(
+      buildRequest({ skipIfActive: false, tags: ["connector:abc"] })
+    );
+
+    expect(result).toEqual({ wasSkipped: false });
+    expect(prisma.$queryRaw).not.toHaveBeenCalled();
+  });
+
+  it("returns wasSkipped=false when no tags are supplied (tag scope required)", async () => {
+    const prisma = mockPrisma();
+    const concern = new SkipIfActiveConcern(prisma as never);
+
+    const result = await concern.handleTriggerRequest(
+      buildRequest({ skipIfActive: true, tags: undefined })
+    );
+
+    expect(result).toEqual({ wasSkipped: false });
+    expect(prisma.$queryRaw).not.toHaveBeenCalled();
+  });
+
+  it("returns wasSkipped=false when no active run matches", async () => {
+    const prisma = mockPrisma({ existing: [] });
+    const concern = new SkipIfActiveConcern(prisma as never);
+
+    const result = await concern.handleTriggerRequest(
+      buildRequest({ skipIfActive: true, tags: ["connector:abc"] })
+    );
+
+    expect(result).toEqual({ wasSkipped: false });
+    expect(prisma.$queryRaw).toHaveBeenCalledTimes(1);
+    expect(prisma.taskRun.findUnique).not.toHaveBeenCalled();
+  });
+
+  it("returns wasSkipped=true with the existing run when an active run matches all tags", async () => {
+    const existingRun = { id: "run_existing", status: "EXECUTING", runTags: ["connector:abc"] };
+    const prisma = mockPrisma({ existing: [{ id: existingRun.id }], run: existingRun });
+    const concern = new SkipIfActiveConcern(prisma as never);
+
+    const result = await concern.handleTriggerRequest(
+      buildRequest({ skipIfActive: true, tags: ["connector:abc"] })
+    );
+
+    expect(result).toMatchObject({ wasSkipped: true, run: existingRun });
+    expect(prisma.$queryRaw).toHaveBeenCalledTimes(1);
+    expect(prisma.taskRun.findUnique).toHaveBeenCalledWith({ where: { id: "run_existing" } });
+  });
+
+  it("normalizes a single tag string into an array", async () => {
+    const prisma = mockPrisma({
+      existing: [{ id: "run_x" }],
+      run: { id: "run_x", status: "PENDING", runTags: ["connector:abc"] },
+    });
+    const concern = new SkipIfActiveConcern(prisma as never);
+
+    const result = await concern.handleTriggerRequest(
+      // @ts-expect-error Zod allows both string and string[] for tags
+      buildRequest({ skipIfActive: true, tags: "connector:abc" })
+    );
+
+    expect(result.wasSkipped).toBe(true);
+  });
+
+  it("recovers gracefully when the row disappears between the probe and the fetch", async () => {
+    const prisma = mockPrisma({ existing: [{ id: "run_gone" }], run: null });
+    const concern = new SkipIfActiveConcern(prisma as never);
+
+    const result = await concern.handleTriggerRequest(
+      buildRequest({ skipIfActive: true, tags: ["connector:abc"] })
+    );
+
+    expect(result).toEqual({ wasSkipped: false });
+  });
+});

docs/docs.json

@@ -70,6 +70,7 @@
               "versioning",
               "machines",
               "idempotency",
+              "skip-if-active",
               "runs/max-duration",
               "tags",
               "runs/metadata",