@@ -0,0 +1,14 @@
+---
+area: webapp
+type: fix
+---
+
+`batchTrigger` now re-triggers when the idempotency key points to a run
+that already ended in a terminal failure state
+(`CRASHED` / `COMPLETED_WITH_ERRORS` / `SYSTEM_FAILURE` / `INTERRUPTED` /
+`TIMED_OUT` / `EXPIRED`), matching the single-trigger behaviour. Previously
+a batch call with the same key as a dead run silently returned the dead
+run as `isCached: true`, leaving the caller waiting on a run that would
+never produce output. Multiple batch items that share the same key now
+also share the same replacement run id, so the response no longer
+contains ghost ids pointing at runs that were never created. Closes #4819.
@@ -27,7 +27,11 @@ import { mintBatchFriendlyId } from "~/v3/runOpsMigration/mintBatchFriendlyId.se
 import { batchTriggerWorker } from "../batchTriggerWorker.server";
 import { guardQueueSizeLimitsForEnv } from "../queueSizeLimits.server";
 import { downloadPacketFromObjectStore, uploadPacketToObjectStore } from "../objectStore.server";
-import { isFinalAttemptStatus, isFinalRunStatus } from "../taskStatus";
+import {
+  isFinalAttemptStatus,
+  isFinalRunStatus,
+  shouldIdempotencyKeyBeCleared,
+} from "../taskStatus";
 import { startActiveSpan } from "../tracer.server";
 import { BaseService, ServiceValidationError } from "./baseService.server";
 import { OutOfEntitlementError, TriggerTaskService } from "./triggerTask.server";
@@ -442,6 +446,10 @@ export class BatchTriggerV3Service extends BaseService {
     // Build the run IDs in order: reuse an unexpired cached id, else mint a new id (and record any
     // expired cached id so its idempotency key can be cleared below).
     const expiredRunIds = new Set<string>();
+    // Dedupes replacement ids when multiple batch items share
+    // (taskIdentifier, idempotencyKey) and the cached run must be cleared.
+    // See the fuller explanation inside the loop below.
+    const replacementIdsByKey = new Map<string, string>();
 
     const runs = await Promise.all(
       body.items.map(async (item) => {
@@ -450,13 +458,55 @@ export class BatchTriggerV3Service extends BaseService {
         );
 
         if (cachedRun) {
-          if (cachedRun.idempotencyKeyExpiresAt && cachedRun.idempotencyKeyExpiresAt < new Date()) {
+          // Clear the idempotency key and mint a fresh run when either:
+          //   (a) the key's time-based expiry has passed, OR
+          //   (b) the previous run ended in a terminal failure state the
+          //       single-trigger path treats as "retry allowed"
+          //       (INTERRUPTED, COMPLETED_WITH_ERRORS, SYSTEM_FAILURE,
+          //       CRASHED, TIMED_OUT, plus EXPIRED — see
+          //       `shouldIdempotencyKeyBeCleared` in `taskStatus.ts`).
+          //
+          // Before this fix, batchTrigger only checked (a). A failed run
+          // with the same idempotency key was returned as `isCached: true`,
+          // silently handing the caller back a dead run that would never
+          // produce output (issue #4819). This mirrors the single-trigger
+          // behaviour at IdempotencyKeyConcern.handleExistingRun
+          // (idempotencyKeys.server.ts:463).
+          const expired =
+            cachedRun.idempotencyKeyExpiresAt &&
+            cachedRun.idempotencyKeyExpiresAt < new Date();
+          const failed = shouldIdempotencyKeyBeCleared(cachedRun.status);
+
+          if (expired || failed) {
             expiredRunIds.add(cachedRun.friendlyId);
 
+            // When two batch items share (task, idempotencyKey) AND the
+            // cached run is cleared, downstream processing creates only ONE
+            // run per key (TriggerTaskService.call honours the key and
+            // reuses the first fresh run for every subsequent item).
+            // Allocating distinct minted ids per item left the response
+            // referencing ghost ids. Reuse the first replacement id for
+            // every item sharing a (task, key). This also fixes the
+            // pre-existing time-expiration branch. (Devin review on #5014.)
+            const idemKey = item.options?.idempotencyKey;
+            let replacementId: string | undefined = idemKey
+              ? replacementIdsByKey.get(`${item.task} ${idemKey}`)
+              : undefined;
+            if (!replacementId) {
+              replacementId = await this.mintChildFriendlyId(
+                environment,
+                childAnchor,
+                item.options?.region
+              );
+              if (idemKey) {
+                replacementIdsByKey.set(`${item.task} ${idemKey}`, replacementId);
+              }
+            }
+
             return {
-              id: await this.mintChildFriendlyId(environment, childAnchor, item.options?.region),
+              id: replacementId,
               isCached: false,
-              idempotencyKey: item.options?.idempotencyKey ?? undefined,
+              idempotencyKey: idemKey ?? undefined,
               taskIdentifier: item.task,
             };
           }
 
@@ -1,5 +1,5 @@
 import { postgresTest } from "@internal/testcontainers";
-import type { PrismaClient } from "@trigger.dev/database";
+import type { PrismaClient, TaskRunStatus } from "@trigger.dev/database";
 import { describe, expect } from "vitest";
 import { PostgresRunStore } from "./PostgresRunStore.js";
 
@@ -38,6 +38,7 @@ async function createRun(
     taskIdentifier: string;
     idempotencyKey: string;
     idempotencyKeyExpiresAt?: Date;
+    status?: TaskRunStatus;
   }
 ) {
   await prisma.taskRun.create({
@@ -46,6 +47,7 @@ async function createRun(
       taskIdentifier: params.taskIdentifier,
       idempotencyKey: params.idempotencyKey,
       idempotencyKeyExpiresAt: params.idempotencyKeyExpiresAt ?? null,
+      status: params.status ?? "PENDING",
       payload: "{}",
       payloadType: "application/json",
       runtimeEnvironmentId: params.runtimeEnvironmentId,
@@ -117,4 +119,62 @@ describe("PostgresRunStore.findRunsByIdempotencyKeys", () => {
 
     expect(rows).toEqual([]);
   });
+
+  // Regression for #4819 — batchTrigger needs the row's status to decide
+  // whether to re-trigger (same semantics as single trigger via
+  // shouldIdempotencyKeyBeCleared). Before this fix the SQL omitted `status`,
+  // so batchTrigger could only inspect time-based expiry and silently handed
+  // callers back dead CRASHED / COMPLETED_WITH_ERRORS runs as `isCached: true`.
+  postgresTest(
+    "returns run status so callers can honour shouldIdempotencyKeyBeCleared",
+    async ({ prisma }) => {
+      const { project, environment } = await seedEnvironment(prisma);
+      const store = new PostgresRunStore({ prisma, readOnlyPrisma: prisma });
+
+      await createRun(prisma, {
+        runtimeEnvironmentId: environment.id,
+        projectId: project.id,
+        friendlyId: "run_pending",
+        taskIdentifier: "task-x",
+        idempotencyKey: "key-pending",
+        status: "PENDING",
+      });
+      await createRun(prisma, {
+        runtimeEnvironmentId: environment.id,
+        projectId: project.id,
+        friendlyId: "run_crashed",
+        taskIdentifier: "task-x",
+        idempotencyKey: "key-crashed",
+        status: "CRASHED",
+      });
+      await createRun(prisma, {
+        runtimeEnvironmentId: environment.id,
+        projectId: project.id,
+        friendlyId: "run_completed_with_errors",
+        taskIdentifier: "task-x",
+        idempotencyKey: "key-cwe",
+        status: "COMPLETED_WITH_ERRORS",
+      });
+      await createRun(prisma, {
+        runtimeEnvironmentId: environment.id,
+        projectId: project.id,
+        friendlyId: "run_completed",
+        taskIdentifier: "task-x",
+        idempotencyKey: "key-ok",
+        status: "COMPLETED_SUCCESSFULLY",
+      });
+
+      const rows = await store.findRunsByIdempotencyKeys({
+        runtimeEnvironmentId: environment.id,
+        taskIdentifier: "task-x",
+        idempotencyKeys: ["key-pending", "key-crashed", "key-cwe", "key-ok"],
+      });
+
+      const byKey = new Map(rows.map((r) => [r.idempotencyKey, r]));
+      expect(byKey.get("key-pending")?.status).toBe("PENDING");
+      expect(byKey.get("key-crashed")?.status).toBe("CRASHED");
+      expect(byKey.get("key-cwe")?.status).toBe("COMPLETED_WITH_ERRORS");
+      expect(byKey.get("key-ok")?.status).toBe("COMPLETED_SUCCESSFULLY");
+    }
+  );
 });
@@ -1924,7 +1924,9 @@ export class PostgresRunStore implements RunStore {
     const branches = args.idempotencyKeys.map((key) => {
       const base = params.length;
       params.push(args.runtimeEnvironmentId, args.taskIdentifier, key);
-      return `SELECT "id", "createdAt", "friendlyId", "idempotencyKey", "idempotencyKeyExpiresAt" FROM "TaskRun" WHERE "runtimeEnvironmentId" = $${base + 1} AND "taskIdentifier" = $${base + 2} AND "idempotencyKey" = $${base + 3}`;
+      // "status" is required so callers can honour shouldIdempotencyKeyBeCleared
+      // (closes #4819 — batchTrigger previously returned stale failed runs as cached).
+      return `SELECT "id", "createdAt", "friendlyId", "idempotencyKey", "idempotencyKeyExpiresAt", "status" FROM "TaskRun" WHERE "runtimeEnvironmentId" = $${base + 1} AND "taskIdentifier" = $${base + 2} AND "idempotencyKey" = $${base + 3}`;
     });
     return prisma.$queryRawUnsafe<IdempotencyKeyRunMatch[]>(
       branches.join(" UNION ALL "),
 
@@ -30,6 +30,14 @@ export type IdempotencyKeyRunMatch = {
   friendlyId: string;
   idempotencyKey: string | null;
   idempotencyKeyExpiresAt: Date | null;
+  /**
+   * Terminal run status is needed so callers can honour the same
+   * "clear the idempotency key if the previous run failed" semantics the
+   * single-trigger path already applies via `shouldIdempotencyKeyBeCleared`.
+   * Without it, `batchTrigger` silently returned stale failed runs as
+   * `isCached: true` (issue #4819).
+   */
+  status: TaskRunStatus;
 };
 
 /**