@@ -4,7 +4,7 @@ import type {
   IOPacket,
 } from "@trigger.dev/core/v3";
 import { packetRequiresOffloading, parsePacket } from "@trigger.dev/core/v3";
-import type { BatchTaskRun, TaskRunAttempt } from "@trigger.dev/database";
+import type { BatchTaskRun, TaskRunAttempt, TaskRunStatus } from "@trigger.dev/database";
 import { isUniqueConstraintError, Prisma } from "@trigger.dev/database";
 import type { RunStore } from "@internal/run-store";
 import pMap from "p-map";
@@ -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";
@@ -461,6 +465,21 @@ export class BatchTriggerV3Service extends BaseService {
             };
           }
 
+          // Mirror the single-trigger path (IdempotencyKeyConcern.handleExistingRun):
+          // if the cached run is in a terminal failure state (CRASHED, SYSTEM_FAILURE,
+          // TIMED_OUT, EXPIRED, COMPLETED_WITH_ERRORS, INTERRUPTED), clear the
+          // idempotency key and re-trigger instead of returning the dead run.
+          if (shouldIdempotencyKeyBeCleared(cachedRun.status as TaskRunStatus)) {
+            expiredRunIds.add(cachedRun.friendlyId);
+
+            return {
+              id: await this.mintChildFriendlyId(environment, childAnchor, item.options?.region),
+              isCached: false,
+              idempotencyKey: item.options?.idempotencyKey ?? undefined,
+              taskIdentifier: item.task,
+            };
+          }
+
           return {
             id: cachedRun.friendlyId,
             isCached: true,
 
@@ -0,0 +1,81 @@
+import { describe, expect, it } from "vitest";
+import type { TaskRunStatus } from "@trigger.dev/database";
+import { shouldIdempotencyKeyBeCleared } from "~/v3/taskStatus";
+
+/**
+ * Tests the shouldIdempotencyKeyBeCleared function which is now used by both the
+ * single-trigger path (IdempotencyKeyConcern.handleExistingRun) and the batch-trigger
+ * path (BatchTriggerV3Service.#prepareRunData).
+ *
+ * Before the fix, the batch path did not call this function at all — it only
+ * checked time-based expiration, silently returning dead runs as cached hits.
+ *
+ * These tests validate:
+ * 1. All failure statuses that should trigger re-triggering
+ * 2. All non-failure statuses that should preserve the cached run
+ * 3. Edge cases around status classification
+ */
+describe("shouldIdempotencyKeyBeCleared — batch trigger parity", () => {
+  // ─── Statuses that MUST clear the idempotency key ──────────────────
+  // These are the statuses where the single-trigger path (and now the
+  // batch-trigger path) clears the key and creates a fresh run.
+
+  const clearableStatuses: TaskRunStatus[] = [
+    "CRASHED",
+    "SYSTEM_FAILURE",
+    "TIMED_OUT",
+    "EXPIRED",
+    "COMPLETED_WITH_ERRORS",
+    "INTERRUPTED",
+  ];
+
+  it.each(clearableStatuses)(
+    "returns true for %s — dead runs must not be returned as cached hits in batch triggers",
+    (status) => {
+      expect(shouldIdempotencyKeyBeCleared(status)).toBe(true);
+    }
+  );
+
+  // ─── Statuses that MUST NOT clear the idempotency key ──────────────
+  // These runs are either still in progress, successfully completed,
+  // or canceled by the user. Returning them as cached is correct.
+
+  const nonClearableStatuses: TaskRunStatus[] = [
+    "PENDING",
+    "PENDING_VERSION",
+    "WAITING_FOR_DEPLOY",
+    "DEQUEUED",
+    "EXECUTING",
+    "WAITING_TO_RESUME",
+    "RETRYING_AFTER_FAILURE",
+    "PAUSED",
+    "DELAYED",
+    "COMPLETED_SUCCESSFULLY",
+    "CANCELED",
+  ];
+
+  it.each(nonClearableStatuses)(
+    "returns false for %s — these runs should be returned as cached hits",
+    (status) => {
+      expect(shouldIdempotencyKeyBeCleared(status)).toBe(false);
+    }
+  );
+
+  // ─── Specific edge cases ───────────────────────────────────────────
+
+  it("COMPLETED_SUCCESSFULLY should NOT be cleared — it is a valid cached result", () => {
+    expect(shouldIdempotencyKeyBeCleared("COMPLETED_SUCCESSFULLY")).toBe(false);
+  });
+
+  it("CANCELED should NOT be cleared — user-initiated cancellation is intentional", () => {
+    expect(shouldIdempotencyKeyBeCleared("CANCELED")).toBe(false);
+  });
+
+  it("EXPIRED should be cleared — expired runs should be re-triggerable via batch", () => {
+    expect(shouldIdempotencyKeyBeCleared("EXPIRED")).toBe(true);
+  });
+
+  it("RETRYING_AFTER_FAILURE should NOT be cleared — the run is still in progress", () => {
+    expect(shouldIdempotencyKeyBeCleared("RETRYING_AFTER_FAILURE")).toBe(false);
+  });
+});
@@ -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,
@@ -103,6 +105,9 @@ describe("PostgresRunStore.findRunsByIdempotencyKeys", () => {
       expiresAt.toISOString()
     );
     expect(byKey.get("idem-2")?.idempotencyKeyExpiresAt).toBeNull();
+    // status must be returned so callers can check shouldIdempotencyKeyBeCleared
+    expect(byKey.get("idem-1")?.status).toBe("PENDING");
+    expect(byKey.get("idem-2")?.status).toBe("PENDING");
   });
 
   postgresTest("short-circuits on an empty key list without querying", async ({ prisma }) => {
@@ -117,4 +122,71 @@ describe("PostgresRunStore.findRunsByIdempotencyKeys", () => {
 
     expect(rows).toEqual([]);
   });
+
+  postgresTest(
+    "returns status for runs in failure states (CRASHED, SYSTEM_FAILURE, TIMED_OUT, EXPIRED, COMPLETED_WITH_ERRORS, INTERRUPTED)",
+    async ({ prisma }) => {
+      const { project, environment } = await seedEnvironment(prisma);
+      const store = new PostgresRunStore({ prisma, readOnlyPrisma: prisma });
+
+      const failureStatuses: TaskRunStatus[] = [
+        "CRASHED",
+        "SYSTEM_FAILURE",
+        "TIMED_OUT",
+        "EXPIRED",
+        "COMPLETED_WITH_ERRORS",
+        "INTERRUPTED",
+      ];
+
+      for (let i = 0; i < failureStatuses.length; i++) {
+        await createRun(prisma, {
+          runtimeEnvironmentId: environment.id,
+          projectId: project.id,
+          friendlyId: `run_fail_${i}`,
+          taskIdentifier: "task-a",
+          idempotencyKey: `idem-fail-${i}`,
+          status: failureStatuses[i],
+        });
+      }
+
+      const rows = await store.findRunsByIdempotencyKeys({
+        runtimeEnvironmentId: environment.id,
+        taskIdentifier: "task-a",
+        idempotencyKeys: failureStatuses.map((_, i) => `idem-fail-${i}`),
+      });
+
+      expect(rows).toHaveLength(failureStatuses.length);
+
+      const byKey = new Map(rows.map((r) => [r.idempotencyKey, r]));
+      for (let i = 0; i < failureStatuses.length; i++) {
+        const row = byKey.get(`idem-fail-${i}`);
+        expect(row).toBeDefined();
+        expect(row?.status).toBe(failureStatuses[i]);
+        expect(row?.friendlyId).toBe(`run_fail_${i}`);
+      }
+    }
+  );
+
+  postgresTest("returns status for a successfully completed run", 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_success",
+      taskIdentifier: "task-a",
+      idempotencyKey: "idem-success",
+      status: "COMPLETED_SUCCESSFULLY",
+    });
+
+    const rows = await store.findRunsByIdempotencyKeys({
+      runtimeEnvironmentId: environment.id,
+      taskIdentifier: "task-a",
+      idempotencyKeys: ["idem-success"],
+    });
+
+    expect(rows).toHaveLength(1);
+    expect(rows[0].status).toBe("COMPLETED_SUCCESSFULLY");
+  });
 });
@@ -1861,7 +1861,7 @@ 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}`;
+      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 "),
 
@@ -28,6 +28,7 @@ export type IdempotencyKeyRunMatch = {
   friendlyId: string;
   idempotencyKey: string | null;
   idempotencyKeyExpiresAt: Date | null;
+  status: string;
 };
 
 export type CreateRunSnapshotInput = {