@@ -78,13 +78,14 @@ import {
   ExecutionSnapshotSystem,
   getExecutionSnapshotsSince,
   getLatestExecutionSnapshot,
+  getLatestExecutionSnapshotOrFinishedRun,
 } from "./systems/executionSnapshotSystem.js";
 import {
   PARKED_ON_EXTERNAL_DEPLOYMENT_STATUS_REASON,
   PendingVersionSystem,
 } from "./systems/pendingVersionSystem.js";
 import { RaceSimulationSystem } from "./systems/raceSimulationSystem.js";
-import { RunAttemptSystem } from "./systems/runAttemptSystem.js";
+import { type CancelRunResult, RunAttemptSystem } from "./systems/runAttemptSystem.js";
 import { NoopPendingVersionRunIdLookup } from "./services/pendingVersionLookup.js";
 import type { SystemResources } from "./systems/systems.js";
 import {
@@ -1753,7 +1754,7 @@ export class RunEngine {
     finalizeRun?: boolean;
     bulkActionId?: string;
     tx?: PrismaClientOrTransaction;
-  }): Promise<ExecutionResult & { alreadyFinished: boolean }> {
+  }): Promise<CancelRunResult> {
     return this.runAttemptSystem.cancelRun({
       runId,
       organizationId,
@@ -2718,13 +2719,35 @@ export class RunEngine {
   }
 
   async #repairRun(runId: string, dryRun: boolean, organizationId: string) {
-    const snapshot = await getLatestExecutionSnapshot(
+    const { snapshot, finishedRun } = await getLatestExecutionSnapshotOrFinishedRun(
       this.prisma,
       runId,
       this.runStore,
-      undefined,
-      organizationId
+      organizationId,
+      { id: true }
     );
+    if (finishedRun) {
+      // The run finished and its snapshot state expired, so any concurrency slot it holds has leaked.
+      this.logger.warn("RunEngine.repairRun: snapshot state expired, run already finished", {
+        runId,
+      });
+
+      if (!dryRun) {
+        await this.worker.enqueueOnce({
+          id: `repair-in-progress-run:${runId}`,
+          job: "repairSnapshot",
+          payload: { runId, snapshotId: null, executionStatus: "FINISHED" },
+          availableAt: new Date(Date.now() + this.repairSnapshotTimeoutMs),
+        });
+      }
+
+      return {
+        action: "repairSnapshot",
+        runId,
+        snapshotStatus: "FINISHED",
+        snapshotId: null,
+      };
+    }
 
     if (
       snapshot.executionStatus === "QUEUED" ||
@@ -3173,13 +3196,57 @@ export class RunEngine {
     executionStatus,
   }: {
     runId: string;
-    snapshotId: string;
+    /** null when scheduled for a finished run whose snapshot state had already expired. */
+    snapshotId: string | null;
     executionStatus: string;
   }) {
     return await this.runLock.lock("handleRepairSnapshot", [runId], async () => {
       // No snapshot mirror in off-only mode, so the repair heals only the queue below. The M1+
       // rebuild reintroduces a mirror-repair step here.
-      const latestSnapshot = await getLatestExecutionSnapshot(this.prisma, runId, this.runStore);
+      const { snapshot: latestSnapshot, finishedRun } =
+        await getLatestExecutionSnapshotOrFinishedRun(
+          this.prisma,
+          runId,
+          this.runStore,
+          undefined,
+          {
+            id: true,
+            queue: true,
+            concurrencyKey: true,
+            runtimeEnvironmentId: true,
+          }
+        );
+      if (finishedRun) {
+        // The run is finished whichever snapshot the job was queued from (a live one that has since
+        // finished and expired, or none when queued after expiry), so its slot has leaked either way.
+        this.logger.warn("RunEngine.handleRepairSnapshot: snapshot state expired, run finished", {
+          runId,
+          snapshotId,
+          repairExecutionStatus: executionStatus,
+        });
+
+        const env = await this.controlPlaneResolver.resolveEnv(finishedRun.runtimeEnvironmentId);
+        if (!env) {
+          this.logger.error("RunEngine.handleRepairSnapshot FINISHED environment not found", {
+            runId,
+          });
+          return;
+        }
+
+        await this.#clearRunFromConcurrencySets({
+          runId,
+          orgId: env.organizationId,
+          queue: finishedRun.queue,
+          concurrencyKey: finishedRun.concurrencyKey ?? undefined,
+          env: {
+            id: env.id,
+            type: env.type,
+            project: { id: env.projectId },
+            organization: { id: env.organizationId },
+          },
+        });
+        return;
+      }
 
       if (latestSnapshot.id !== snapshotId) {
         this.logger.log(
@@ -3264,8 +3331,7 @@ export class RunEngine {
             return;
           }
 
-          // We need to clear this run from the current concurrency sets
-          await this.runQueue.clearMessageFromConcurrencySets({
+          await this.#clearRunFromConcurrencySets({
             runId,
             orgId: latestSnapshot.organizationId,
             queue: taskRun.queue,
@@ -3291,6 +3357,13 @@ export class RunEngine {
     });
   }
 
+  async #clearRunFromConcurrencySets(
+    params: Parameters<RunQueue["clearMessageFromConcurrencySets"]>[0]
+  ) {
+    // We need to clear this run from the current concurrency sets
+    await this.runQueue.clearMessageFromConcurrencySets(params);
+  }
+
   /**
    * Applies `defaultMaxTtl` to a run's TTL:
    * - No max configured → pass through as-is.
 
@@ -24,8 +24,10 @@ import { getMachinePreset } from "../machinePresets.js";
 import { isDequeueableExecutionStatus, isExecuting } from "../statuses.js";
 import type { WorkerQueueSelection } from "../../run-queue/types.js";
 import type { RunEngineOptions, WorkerQueueDequeueOptions } from "../types.js";
-import type { ExecutionSnapshotSystem } from "./executionSnapshotSystem.js";
-import { getLatestExecutionSnapshot } from "./executionSnapshotSystem.js";
+import {
+  getLatestExecutionSnapshotOrFinishedRun,
+  type ExecutionSnapshotSystem,
+} from "./executionSnapshotSystem.js";
 import type { RunAttemptSystem } from "./runAttemptSystem.js";
 import type { SystemResources } from "./systems.js";
 import { snapshotRouteFromSnapshot, type SnapshotRouteWire } from "@internal/run-store";
@@ -197,13 +199,18 @@ export class DequeueSystem {
             "dequeueFromWorkerQueue",
             [runId],
             async () => {
-              const snapshot = await getLatestExecutionSnapshot(
+              const { snapshot, finishedRun } = await getLatestExecutionSnapshotOrFinishedRun(
                 prisma,
                 runId,
                 this.$.runStore,
-                undefined,
-                orgId
+                orgId,
+                { id: true }
               );
+              if (finishedRun) {
+                // A stale message for a finished run: remove it instead of starting the run.
+                await this.#acknowledgeExpiredFinishedRun(orgId, runId);
+                return;
+              }
               preparedFromSnapshotId = snapshot.id;
 
               if (!isDequeueableExecutionStatus(snapshot.executionStatus)) {
@@ -758,11 +765,14 @@ export class DequeueSystem {
           // Reentrant: tryNackAndRequeue's own lock on this run nests inside this one.
           await this.$.runLock.lock("dequeueFromWorkerQueueRecovery", [runId], async () => {
             // Preparation may have committed a transition before throwing, so read current state.
-            const [snapshotError, snapshot] = await tryCatch(
-              getLatestExecutionSnapshot(prisma, runId, this.$.runStore)
+            // A finished run's stale message is removed; any read error (including a failed row
+            // read after expiry) falls through to the nack below.
+            const [snapshotError, read] = await tryCatch(
+              getLatestExecutionSnapshotOrFinishedRun(prisma, runId, this.$.runStore, undefined, {
+                id: true,
+              })
             );
-
-            if (snapshotError) {
+            if (snapshotError || !read) {
               // Do not replace execution state we could not read with a checkpoint-less snapshot.
               const requeued = await this.$.runQueue.nackMessage({ orgId, messageId: runId });
               if (requeued === false) {
@@ -775,6 +785,11 @@ export class DequeueSystem {
               }
               return;
             }
+            if (read.finishedRun) {
+              await this.#acknowledgeExpiredFinishedRun(orgId, runId);
+              return;
+            }
+            const snapshot = read.snapshot;
 
             // A PENDING_EXECUTING snapshot written by this preparation was never handed to a worker,
             // so it is still ours to requeue. Any other non-dequeueable state is owned elsewhere.
@@ -847,6 +862,14 @@ export class DequeueSystem {
     );
   }
 
+  async #acknowledgeExpiredFinishedRun(orgId: string, runId: string): Promise<void> {
+    this.$.logger.warn(
+      "dequeueFromWorkerQueue: snapshot state expired for a finished run, removing from queue",
+      { runId, orgId }
+    );
+    await this.$.runQueue.acknowledgeMessage(orgId, runId);
+  }
+
   async #pendingVersion({
     orgId,
     runId,
 
@@ -18,8 +18,9 @@ import type {
   SnapshotReadContext,
   SnapshotRouteWire,
 } from "@internal/run-store";
-import { snapshotRouteFromSnapshot } from "@internal/run-store";
+import { SnapshotStateExpiredError, snapshotRouteFromSnapshot } from "@internal/run-store";
 import { ExecutionSnapshotNotFoundError, ServiceValidationError } from "../errors.js";
+import { isFinalRunStatus } from "../statuses.js";
 import type { HeartbeatTimeouts } from "../types.js";
 import type { SystemResources } from "./systems.js";
 
@@ -213,6 +214,62 @@ async function fetchWaitpointsInChunks(
   return allWaitpoints;
 }
 
+/**
+ * True when a snapshot read failed because a redis-primary run's state aged out and the run row is
+ * final. Snapshot state only gets a TTL from a FINISHED entry, so a final row means the run finished.
+ * Expired state on a non-final row is lost state and must keep failing closed.
+ */
+export function isExpiredSnapshotOfFinishedRun(error: unknown, runStatus: TaskRunStatus): boolean {
+  return error instanceof SnapshotStateExpiredError && isFinalRunStatus(runStatus);
+}
+
+type FinishedRunRow<S extends Prisma.TaskRunSelect> = Prisma.TaskRunGetPayload<{
+  select: S;
+}> & { status: TaskRunStatus };
+
+export type LatestExecutionSnapshotOrFinishedRun<S extends Prisma.TaskRunSelect> =
+  | { snapshot: EnhancedExecutionSnapshot; finishedRun: null }
+  | { snapshot: null; finishedRun: FinishedRunRow<S> };
+
+/**
+ * Like `getLatestExecutionSnapshot`, but when the run's snapshot state has expired and its run row
+ * is final, returns that row (selected with `runSelect` plus `status`) instead of throwing. Every
+ * other error, and expired state on a non-final row, is rethrown unchanged.
+ */
+export async function getLatestExecutionSnapshotOrFinishedRun<S extends Prisma.TaskRunSelect>(
+  prisma: PrismaClientOrTransaction,
+  runId: string,
+  runStore: RunStore,
+  organizationId: SnapshotReadContext | undefined,
+  runSelect: S
+): Promise<LatestExecutionSnapshotOrFinishedRun<S>> {
+  try {
+    const snapshot = await getLatestExecutionSnapshot(
+      prisma,
+      runId,
+      runStore,
+      undefined,
+      organizationId
+    );
+    return { snapshot, finishedRun: null };
+  } catch (error) {
+    if (!(error instanceof SnapshotStateExpiredError)) {
+      throw error;
+    }
+    // Prisma's select-payload mapping does not compose over an intersection, so the row is read with
+    // `status` added and typed as the caller's selection plus `status`.
+    const run = (await runStore.findRun(
+      { id: runId },
+      { select: { ...runSelect, status: true } },
+      prisma
+    )) as FinishedRunRow<S> | null;
+    if (!run || !isExpiredSnapshotOfFinishedRun(error, run.status)) {
+      throw error;
+    }
+    return { snapshot: null, finishedRun: run };
+  }
+}
+
 /**
  * Gets the most recent valid snapshot for a run. When `environmentId` is provided the read is scoped
  * to that environment (tenant boundary): a run in another environment reads as not-found and rejects
 
@@ -66,12 +66,18 @@ import type {
 import {
   executionResultFromSnapshot,
   getLatestExecutionSnapshot,
+  getLatestExecutionSnapshotOrFinishedRun,
 } from "./executionSnapshotSystem.js";
 import type { SystemResources } from "./systems.js";
 import { snapshotRouteFromSnapshot, type SnapshotRouteWire } from "@internal/run-store";
 import type { WaitpointSystem } from "./waitpointSystem.js";
 import { BatchId, RunId } from "@trigger.dev/core/v3/isomorphic";
 
+/** A run whose snapshot state has expired has no snapshot to report, only its final run row. */
+export type CancelRunResult =
+  | (ExecutionResult & { alreadyFinished: boolean })
+  | { alreadyFinished: true; run: ExecutionResult["run"]; snapshot: null };
+
 export type RunAttemptSystemOptions = {
   resources: SystemResources;
   executionSnapshotSystem: ExecutionSnapshotSystem;
@@ -1060,6 +1066,12 @@ export class RunAttemptSystem {
                 attemptDurationMs: completion.usage?.durationMs,
                 tx: prisma,
               });
+              if (result.snapshot === null) {
+                // Not reachable for a run with a live attempt.
+                throw new Error(
+                  `cancelRun returned no snapshot for run ${runId} with a live attempt`
+                );
+              }
               return {
                 attemptStatus:
                   result.snapshot.executionStatus === "PENDING_CANCEL"
@@ -1604,18 +1616,26 @@ export class RunAttemptSystem {
     bulkActionId?: string;
     attemptDurationMs?: number;
     tx?: PrismaClientOrTransaction;
-  }): Promise<ExecutionResult & { alreadyFinished: boolean }> {
+  }): Promise<CancelRunResult> {
     const prisma = tx ?? this.$.prisma;
 
     return startSpan(this.$.tracer, "cancelRun", async (span) => {
-      return this.$.runLock.lock("cancelRun", [runId], async () => {
-        const latestSnapshot = await getLatestExecutionSnapshot(
-          prisma,
-          runId,
-          this.$.runStore,
-          undefined,
-          organizationId
-        );
+      return this.$.runLock.lock("cancelRun", [runId], async (): Promise<CancelRunResult> => {
+        const { snapshot: latestSnapshot, finishedRun } =
+          await getLatestExecutionSnapshotOrFinishedRun(
+            prisma,
+            runId,
+            this.$.runStore,
+            organizationId,
+            { id: true, friendlyId: true, attemptNumber: true, taskEventStore: true }
+          );
+        if (finishedRun) {
+          if (bulkActionId) {
+            await this.$.runStore.recordBulkActionMembership(runId, bulkActionId, prisma);
+          }
+          this.$.logger.warn("cancelRun: snapshot state expired, run already finished", { runId });
+          return { alreadyFinished: true, run: finishedRun, snapshot: null };
+        }
 
         //already finished, do nothing
         if (latestSnapshot.executionStatus === "FINISHED") {