fix(run-engine): correct park deadline and snapshot state for debounced parked runs

Two defects surface when a run parked on an external deployment id is pushed by
a debounce key. Both were reproduced against a local instance before fixing.

1. The run is expired before it is due.

The park deadline is armed once, when the run is first parked, from
max(now, delayUntil) + deadline. Debounce pushes delayUntil out afterwards:
rescheduleDelayedRun reschedules enqueueDelayedRun:<id>, and the redis-worker
reschedule is an update-only ZADD, so expireParkedExternalDeploymentRun:<id> is
never re-armed. Repeat triggers on one key walk delayUntil past a deadline that
no longer moves, and the run is expired with EXTERNAL_DEPLOYMENT_NOT_FOUND
before it was ever due to start. Observed: a run due at 14:01:37 expired at
13:57:02.

The expiry job already loads delayUntil, so it now re-arms from the current
value and returns instead of expiring a run that is not due. Putting the guard
there rather than in the debounce path covers every caller that moves
delayUntil, and it stays bounded by the debounce max-duration contract.

2. The run reports itself as delayed while it is parked.

rescheduleRun hardcoded a DELAYED/DELAYED execution snapshot, so a debounce push
left the run row on PENDING_VERSION while its latest snapshot claimed DELAYED,
and the run page described a parked run as delayed. The snapshot statuses are
now supplied by the caller and default to DELAYED, so the ordinary delayed path
is unchanged, and rescheduleDelayedRun passes the parked statuses through when
the run is parked.

Author: 0ski

internal-packages/run-engine/src/engine/systems/delayedRunSystem.ts

@@ -48,6 +48,8 @@ export class DelayedRunSystem {
             throw new ServiceValidationError("Cannot reschedule a run that is not delayed");
           }
 
+          const isParked = snapshot.runStatus === "PENDING_VERSION";
+
           const updatedRun = await this.$.runStore.rescheduleRun(
             runId,
             {
@@ -57,6 +59,13 @@ export class DelayedRunSystem {
                 environmentType: snapshot.environmentType,
                 projectId: snapshot.projectId,
                 organizationId: snapshot.organizationId,
+                ...(isParked
+                  ? {
+                      executionStatus: "RUN_CREATED" as const,
+                      runStatus: "PENDING_VERSION" as const,
+                      description: "Parked run was rescheduled to a future date",
+                    }
+                  : {}),
               },
             },
             prisma

internal-packages/run-engine/src/engine/systems/pendingVersionSystem.ts

@@ -470,6 +470,22 @@ export class PendingVersionSystem {
       );
     }
 
+    if (run.delayUntil && run.delayUntil > new Date()) {
+      this.$.logger.info(
+        "expireParkedExternalDeploymentRun: run is not due yet, re-arming the park deadline",
+        { runId, externalDeploymentId, delayUntil: run.delayUntil }
+      );
+
+      await this.scheduleExternalDeploymentParkDeadline({
+        runId,
+        externalDeploymentId,
+        ttl: run.ttl,
+        delayUntil: run.delayUntil,
+      });
+
+      return;
+    }
+
     const error: TaskRunError = {
       type: "STRING_ERROR",
       raw: `Run expired because no deployment with external id '${externalDeploymentId}' became available`,

internal-packages/run-engine/src/engine/tests/externalDeploymentParking.test.ts

@@ -566,6 +566,128 @@ describe("RunEngine external deployment parking", () => {
     }
   );
 
+  containerTest(
+    "a debounce push on a parked run keeps the snapshot parked instead of reporting it delayed",
+    async ({ prisma, redisOptions }) => {
+      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
+      const engine = createEngine(prisma, redisOptions);
+
+      try {
+        const taskIdentifier = "test-task";
+
+        const run = await engine.trigger(
+          {
+            number: 1,
+            friendlyId: "run_1234",
+            environment: authenticatedEnvironment,
+            taskIdentifier,
+            payload: "{}",
+            payloadType: "application/json",
+            context: {},
+            traceContext: {},
+            traceId: "t1234",
+            spanId: "s1234",
+            queue: `task/${taskIdentifier}`,
+            isTest: false,
+            tags: [],
+            delayUntil: new Date(Date.now() + 60 * 1000),
+            annotations: {
+              triggerSource: "sdk",
+              triggerAction: "trigger",
+              rootTriggerSource: "sdk",
+              externalDeploymentId: "commit-snapshot",
+            },
+            parkedOnExternalDeploymentId: "commit-snapshot",
+          },
+          prisma
+        );
+
+        await engine.delayedRunSystem.rescheduleDelayedRun({
+          runId: run.id,
+          delayUntil: new Date(Date.now() + 10 * 60 * 1000),
+          tx: prisma,
+        });
+
+        const snapshots = await prisma.taskRunExecutionSnapshot.findMany({
+          where: { runId: run.id },
+          orderBy: { createdAt: "asc" },
+          select: { executionStatus: true, runStatus: true },
+        });
+
+        const latest = snapshots.at(-1);
+
+        assertNonNullable(latest);
+        expect(latest.runStatus).toBe("PENDING_VERSION");
+        expect(latest.executionStatus).toBe("RUN_CREATED");
+
+        const stillParked = await prisma.taskRun.findFirstOrThrow({ where: { id: run.id } });
+        expect(stillParked.status).toBe("PENDING_VERSION");
+      } finally {
+        await engine.quit();
+      }
+    }
+  );
+
+  containerTest(
+    "the parking deadline re-arms instead of expiring a run whose delay was pushed past it",
+    async ({ prisma, redisOptions }) => {
+      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
+      const engine = createEngine(prisma, redisOptions);
+
+      try {
+        const taskIdentifier = "test-task";
+
+        const run = await engine.trigger(
+          {
+            number: 1,
+            friendlyId: "run_1234",
+            environment: authenticatedEnvironment,
+            taskIdentifier,
+            payload: "{}",
+            payloadType: "application/json",
+            context: {},
+            traceContext: {},
+            traceId: "t1234",
+            spanId: "s1234",
+            queue: `task/${taskIdentifier}`,
+            isTest: false,
+            tags: [],
+            delayUntil: new Date(Date.now() + 60 * 1000),
+            annotations: {
+              triggerSource: "sdk",
+              triggerAction: "trigger",
+              rootTriggerSource: "sdk",
+              externalDeploymentId: "commit-pushed",
+            },
+            parkedOnExternalDeploymentId: "commit-pushed",
+          },
+          prisma
+        );
+
+        // Stand in for a debounce push: the run's delay moves out, but nothing re-arms the
+        // deadline that was computed when the run was first parked.
+        const pushedDelayUntil = new Date(Date.now() + 60 * 60 * 1000);
+        await prisma.taskRun.update({
+          where: { id: run.id },
+          data: { delayUntil: pushedDelayUntil },
+        });
+
+        await engine.pendingVersionSystem.expireParkedExternalDeploymentRun({
+          runId: run.id,
+          externalDeploymentId: "commit-pushed",
+        });
+
+        const stillParked = await prisma.taskRun.findFirstOrThrow({ where: { id: run.id } });
+
+        expect(stillParked.status).toBe("PENDING_VERSION");
+        expect(stillParked.statusReason).toBe("EXTERNAL_DEPLOYMENT_PENDING");
+        expect(stillParked.expiredAt).toBeNull();
+      } finally {
+        await engine.quit();
+      }
+    }
+  );
+
   containerTest(
     "the parking deadline expires a run whose deployment never arrived",
     async ({ prisma, redisOptions }) => {

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

@@ -1439,9 +1439,10 @@ export class PostgresRunStore implements RunStore {
           executionSnapshots: {
             create: {
               engine: "V2",
-              executionStatus: "DELAYED",
-              description: "Delayed run was rescheduled to a future date",
-              runStatus: "DELAYED",
+              executionStatus: data.snapshot.executionStatus ?? "DELAYED",
+              description:
+                data.snapshot.description ?? "Delayed run was rescheduled to a future date",
+              runStatus: data.snapshot.runStatus ?? "DELAYED",
               environmentId: data.snapshot.environmentId,
               environmentType: data.snapshot.environmentType,
               projectId: data.snapshot.projectId,

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

@@ -79,6 +79,9 @@ export type RescheduleSnapshotInput = {
   environmentType: RuntimeEnvironmentType;
   projectId: string;
   organizationId: string;
+  executionStatus?: TaskRunExecutionStatus;
+  runStatus?: TaskRunStatus;
+  description?: string;
 };
 
 export type LockSnapshotInput = {