fix(batch): clear idempotency keys pointing at dead runs in batchTrigger

The single-trigger path already clears an idempotency key when the
cached run reached a clearable terminal state
(shouldIdempotencyKeyBeCleared: failed statuses + EXPIRED). The batch
path only tested time expiry, so batchTrigger with an idempotency key
pointing at a FAILED run returned the dead run as isCached forever.

- PostgresRunStore.findRunsByIdempotencyKeys now selects the status
  column
- IdempotencyKeyRunMatch gains status
- batchTriggerV3 clears the key and mints a fresh run when
  shouldIdempotencyKeyBeCleared(status) holds, same branch as time
  expiry

Credit: reported by @Jaimin2687 in #4819; fix shape follows
@itzzdev09's analysis in the issue thread.

Author: Instinct Agent

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

@@ -27,7 +27,7 @@ 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";
@@ -450,7 +450,13 @@ export class BatchTriggerV3Service extends BaseService {
         );
 
         if (cachedRun) {
-          if (cachedRun.idempotencyKeyExpiresAt && cachedRun.idempotencyKeyExpiresAt < new Date()) {
+          // Parity with the single-trigger path (IdempotencyKeyConcern.handleExistingRun): a cached
+          // run that reached a clearable terminal state (failed/expired) must not be returned as
+          // isCached - clear the key and mint a fresh run, same as a time-expired one.
+          if (
+            (cachedRun.idempotencyKeyExpiresAt && cachedRun.idempotencyKeyExpiresAt < new Date()) ||
+            shouldIdempotencyKeyBeCleared(cachedRun.status)
+          ) {
             expiredRunIds.add(cachedRun.friendlyId);
 
             return {

internal-packages/run-store/src/PostgresRunStore.findRunsByIdempotencyKeys.test.ts

@@ -105,6 +105,42 @@ describe("PostgresRunStore.findRunsByIdempotencyKeys", () => {
     expect(byKey.get("idem-2")?.idempotencyKeyExpiresAt).toBeNull();
   });
 
+  postgresTest("returns run status so callers can reject dead cached runs", 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_crashed1",
+      taskIdentifier: "task-a",
+      idempotencyKey: "idem-dead",
+    });
+    await prisma.taskRun.update({
+      where: { friendlyId: "run_crashed1" },
+      data: { status: "CRASHED" },
+    });
+    await createRun(prisma, {
+      runtimeEnvironmentId: environment.id,
+      projectId: project.id,
+      friendlyId: "run_ok1",
+      taskIdentifier: "task-a",
+      idempotencyKey: "idem-live",
+    });
+
+    const rows = await store.findRunsByIdempotencyKeys({
+      runtimeEnvironmentId: environment.id,
+      taskIdentifier: "task-a",
+      idempotencyKeys: ["idem-dead", "idem-live"],
+    });
+
+    const byKey = new Map(rows.map((r) => [r.idempotencyKey, r]));
+    // batchTriggerV3 clears the key and mints a fresh run when shouldIdempotencyKeyBeCleared(status)
+    // holds - it can only do that if the lookup actually returns the status column.
+    expect(byKey.get("idem-dead")?.status).toBe("CRASHED");
+    expect(byKey.get("idem-live")?.status).toBe("PENDING");
+  });
+
   postgresTest("short-circuits on an empty key list without querying", async ({ prisma }) => {
     const { environment } = await seedEnvironment(prisma);
     const store = new PostgresRunStore({ prisma, readOnlyPrisma: prisma });

internal-packages/run-store/src/PostgresRunStore.ts

@@ -1916,7 +1916,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 "),

internal-packages/run-store/src/types.ts

@@ -30,6 +30,7 @@ export type IdempotencyKeyRunMatch = {
   friendlyId: string;
   idempotencyKey: string | null;
   idempotencyKeyExpiresAt: Date | null;
+  status: TaskRunStatus;
 };
 
 /**