fix(execution): give a cancelled async run its terminal metadata

apps/sim/app/api/cron/cleanup-stale-executions/route.test.ts

@@ -37,6 +37,21 @@ function flattenConditions(condition: unknown): MockCondition[] {
   return [node, ...(node.conditions?.flatMap((child) => flattenConditions(child)) ?? [])]
 }
 
+function hasToSQL(value: unknown): value is { toSQL: () => { sql: string; params: unknown[] } } {
+  return typeof value === 'object' && value !== null && 'toSQL' in value
+}
+
+/**
+ * Collects the leaves of a nested `sql` expression. The duration expression is
+ * built by a shared helper, so the values it binds sit one level below the
+ * fragment this route assembles rather than directly in its own params.
+ */
+function flattenSqlParams(expression: { sql: string; params: unknown[] }): unknown[] {
+  return expression.params.flatMap((param) =>
+    hasToSQL(param) ? flattenSqlParams(param.toSQL()) : [param]
+  )
+}
+
 function createRequest() {
   return createMockRequest(
     'GET',
@@ -96,11 +111,7 @@ describe('stale execution cleanup deadline grace', () => {
           'toSQL' in value &&
           value.toSQL().sql.includes('EXTRACT(EPOCH')
       )
-      const totalDurationExpression = update.totalDurationMs.toSQL()
-      const cleanupTimestamp = totalDurationExpression.params.find(
-        (value): value is { toSQL: () => { sql: string; params: unknown[] } } =>
-          typeof value === 'object' && value !== null && 'toSQL' in value
-      )
+      const totalDurationLeaves = flattenSqlParams(update.totalDurationMs.toSQL())
 
       expect(errorExpression.sql).toContain('CASE')
       expect(errorExpression.sql).toContain('IS NOT NULL')
@@ -111,11 +122,9 @@ describe('stale execution cleanup deadline grace', () => {
       )
       expect(staleDurationExpression?.toSQL().sql).toContain('ROUND')
       expect(staleDurationExpression?.toSQL().params).toContain(workflowExecutionLogs.startedAt)
-      expect(totalDurationExpression.sql).toContain('LEAST')
-      expect(totalDurationExpression.sql).toContain('ROUND')
-      expect(totalDurationExpression.params).toContain(2_147_483_647)
-      expect(totalDurationExpression.params).toContain(workflowExecutionLogs.startedAt)
-      expect(cleanupTimestamp?.toSQL().params).toEqual([new Date('2026-08-03T12:10:00.000Z')])
+      expect(totalDurationLeaves).toContain(2_147_483_647)
+      expect(totalDurationLeaves).toContain(workflowExecutionLogs.startedAt)
+      expect(totalDurationLeaves).toContainEqual(new Date('2026-08-03T12:10:00.000Z'))
       expect(update.endedAt).toEqual(new Date('2026-08-03T12:10:00.000Z'))
     } finally {
       vi.useRealTimers()

apps/sim/app/api/cron/cleanup-stale-executions/route.ts

@@ -25,6 +25,7 @@ import {
 } from '@/lib/core/execution-limits'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import type { DbTransaction } from '@/lib/db/types'
+import { elapsedDurationMsSql } from '@/lib/logs/execution/duration'
 import { deleteFile } from '@/lib/uploads/core/storage-service'
 
 const logger = createLogger('CleanupStaleExecutions')
@@ -33,7 +34,6 @@ const STALE_THRESHOLD_MS = getExecutionReservationTtlMs()
 const STALE_THRESHOLD_MINUTES = Math.ceil(STALE_THRESHOLD_MS / 60000)
 const GENERIC_STALE_PROCESSING_ERROR = `Job terminated: stuck in processing for more than ${STALE_THRESHOLD_MINUTES} minutes`
 const EXECUTION_DEADLINE_ERROR = getTimeoutErrorMessage(undefined)
-const MAX_INT32 = 2_147_483_647
 /**
  * Table jobs run as detached workers with progress heartbeats, independently of workflow timeout
  * policy. Preserve their historical 90-minute task window plus five-minute cleanup grace.
@@ -154,10 +154,7 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       const staleDurationMinutes = sql<number>`ROUND(
         EXTRACT(EPOCH FROM (${cleanupTimestamp} - ${workflowExecutionLogs.startedAt})) / 60
       )::integer`
-      const totalDurationMs = sql<number>`LEAST(
-        ${MAX_INT32},
-        ROUND(EXTRACT(EPOCH FROM (${cleanupTimestamp} - ${workflowExecutionLogs.startedAt})) * 1000)
-      )::integer`
+      const totalDurationMs = elapsedDurationMsSql(now)
       let workflowRowsConsidered = 0
       while (workflowRowsConsidered < WORKFLOW_EXECUTION_MAX_ROWS_PER_RUN) {
         const limit = Math.min(

apps/sim/app/api/workflows/[id]/executions/[executionId]/cancel/route.test.ts

@@ -1320,6 +1320,7 @@ describe('POST /api/workflows/[id]/executions/[executionId]/cancel', () => {
     expect(mockSet).toHaveBeenCalledWith({
       status: 'cancelled',
       endedAt: expect.any(Date),
+      totalDurationMs: expect.anything(),
       executionDeadlineAt: null,
     })
   })
@@ -1340,6 +1341,7 @@ describe('POST /api/workflows/[id]/executions/[executionId]/cancel', () => {
     expect(mockSet).toHaveBeenCalledWith({
       status: 'cancelled',
       endedAt: expect.any(Date),
+      totalDurationMs: expect.anything(),
       executionDeadlineAt: null,
     })
   })

apps/sim/app/api/workflows/[id]/executions/[executionId]/cancel/route.ts

@@ -24,6 +24,7 @@ import {
 } from '@/lib/execution/cancellation'
 import { createExecutionEventWriter, readExecutionMetaState } from '@/lib/execution/event-buffer'
 import { abortManualExecution } from '@/lib/execution/manual-cancellation'
+import { cancelledExecutionLogFields } from '@/lib/logs/execution/cancellation'
 import { workflowExecutionOriginSql } from '@/lib/logs/execution-origin'
 import { captureServerEvent } from '@/lib/posthog/server'
 import {
@@ -215,7 +216,7 @@ async function claimExecutionLogCancellation(args: {
   const now = new Date()
   const [cancelledExecution] = await db
     .update(workflowExecutionLogs)
-    .set({ status: 'cancelled', endedAt: now, executionDeadlineAt: null })
+    .set(cancelledExecutionLogFields(now))
     .where(
       and(
         eq(workflowExecutionLogs.executionId, args.executionId),

apps/sim/lib/core/async-jobs/backends/trigger-dev.test.ts

@@ -331,6 +331,80 @@ describe('TriggerDevJobQueue status mapping', () => {
 
     await expect(queue.getJob('run-1')).resolves.toMatchObject({ status: jobStatus })
   })
+
+  it('dates a run cancelled before it was dequeued by its last transition', async () => {
+    mockRetrieve.mockResolvedValueOnce({
+      id: 'run-1',
+      payload: {},
+      status: 'CANCELED',
+      taskIdentifier: 'workflow-execution',
+      createdAt: new Date('2026-08-05T12:00:00.000Z'),
+      updatedAt: new Date('2026-08-05T12:00:02.000Z'),
+    })
+    const queue = new TriggerDevJobQueue()
+
+    await expect(queue.getJob('run-1')).resolves.toMatchObject({
+      status: 'cancelled',
+      startedAt: undefined,
+      completedAt: new Date('2026-08-05T12:00:02.000Z'),
+    })
+  })
+
+  it('dates a run cancelled mid-flight by its last transition until it drains', async () => {
+    mockRetrieve.mockResolvedValueOnce({
+      id: 'run-1',
+      payload: {},
+      status: 'CANCELED',
+      taskIdentifier: 'workflow-execution',
+      createdAt: new Date('2026-08-05T12:00:00.000Z'),
+      startedAt: new Date('2026-08-05T12:00:01.000Z'),
+      updatedAt: new Date('2026-08-05T12:00:03.000Z'),
+    })
+    const queue = new TriggerDevJobQueue()
+
+    await expect(queue.getJob('run-1')).resolves.toMatchObject({
+      status: 'cancelled',
+      startedAt: new Date('2026-08-05T12:00:01.000Z'),
+      completedAt: new Date('2026-08-05T12:00:03.000Z'),
+    })
+  })
+
+  it('prefers the reported finish over the last transition once the run has drained', async () => {
+    mockRetrieve.mockResolvedValueOnce({
+      id: 'run-1',
+      payload: {},
+      status: 'COMPLETED',
+      taskIdentifier: 'workflow-execution',
+      createdAt: new Date('2026-08-05T12:00:00.000Z'),
+      startedAt: new Date('2026-08-05T12:00:01.000Z'),
+      finishedAt: new Date('2026-08-05T12:00:04.000Z'),
+      updatedAt: new Date('2026-08-05T12:00:09.000Z'),
+    })
+    const queue = new TriggerDevJobQueue()
+
+    await expect(queue.getJob('run-1')).resolves.toMatchObject({
+      status: 'completed',
+      completedAt: new Date('2026-08-05T12:00:04.000Z'),
+    })
+  })
+
+  it('leaves a still-running job with no completion instant', async () => {
+    mockRetrieve.mockResolvedValueOnce({
+      id: 'run-1',
+      payload: {},
+      status: 'EXECUTING',
+      taskIdentifier: 'workflow-execution',
+      createdAt: new Date('2026-08-05T12:00:00.000Z'),
+      startedAt: new Date('2026-08-05T12:00:01.000Z'),
+      updatedAt: new Date('2026-08-05T12:00:03.000Z'),
+    })
+    const queue = new TriggerDevJobQueue()
+
+    await expect(queue.getJob('run-1')).resolves.toMatchObject({
+      status: 'processing',
+      completedAt: undefined,
+    })
+  })
 })
 
 describe('TriggerDevJobQueue cancellation', () => {

apps/sim/lib/core/async-jobs/backends/trigger-dev.ts

@@ -17,6 +17,7 @@ import {
   type JobQueueBackend,
   type JobStatus,
   type JobType,
+  TERMINAL_JOB_STATUSES,
   validateMaxDurationSeconds,
 } from '@/lib/core/async-jobs/types'
 import { recordExecutionCancellationBackendResult } from '@/lib/core/execution-limits/metrics'
@@ -213,6 +214,32 @@ function mapTriggerDevStatus(status: string): JobStatus {
   }
 }
 
+/**
+ * Dates the end of a run that trigger.dev already reports as terminal.
+ *
+ * A cancellation flips the run to `CANCELED` the moment it is accepted, but
+ * `finishedAt` is only stamped once the worker actually drains — seconds later,
+ * or never for a run that was cancelled before it was ever dequeued. A reader
+ * polling inside that window sees a terminal job carrying no completion
+ * instant, and every consumer that derives an end timestamp or an elapsed
+ * duration from it reports null.
+ *
+ * `updatedAt` is trigger.dev's own record of when the run last changed, so for
+ * a terminal run it dates that final transition rather than the read. It is
+ * consulted only once the status is terminal: an active run's `updatedAt`
+ * describes progress, not an ending, and reporting it would end a run that is
+ * still going.
+ */
+function resolveRunCompletedAt(
+  finishedAt: Date | string | undefined,
+  updatedAt: Date | string | undefined,
+  status: JobStatus
+): Date | undefined {
+  if (finishedAt) return new Date(finishedAt)
+  if (!TERMINAL_JOB_STATUSES.includes(status)) return undefined
+  return updatedAt ? new Date(updatedAt) : undefined
+}
+
 /**
  * Adapter that wraps the trigger.dev SDK to conform to JobQueueBackend interface.
  */
@@ -379,14 +406,16 @@ export class TriggerDevJobQueue implements JobQueueBackend {
             : undefined,
       }
 
+      const status = mapTriggerDevStatus(run.status)
+
       return {
         id: run.id,
         type: run.taskIdentifier as JobType,
         payload: run.payload,
-        status: mapTriggerDevStatus(run.status),
+        status,
         createdAt: run.createdAt ? new Date(run.createdAt) : new Date(),
         startedAt: run.startedAt ? new Date(run.startedAt) : undefined,
-        completedAt: run.finishedAt ? new Date(run.finishedAt) : undefined,
+        completedAt: resolveRunCompletedAt(run.finishedAt, run.updatedAt, status),
         attempts: run.attemptCount ?? 1,
         maxAttempts: 3,
         error: run.error?.message,

apps/sim/lib/core/async-jobs/types.ts

@@ -29,6 +29,13 @@ export const JOB_STATUS = {
 
 export type JobStatus = (typeof JOB_STATUS)[keyof typeof JOB_STATUS]
 
+/** The statuses a job cannot leave; every one of them requires a `completedAt`. */
+export const TERMINAL_JOB_STATUSES: readonly JobStatus[] = [
+  JOB_STATUS.COMPLETED,
+  JOB_STATUS.FAILED,
+  JOB_STATUS.CANCELLED,
+]
+
 export type JobType =
   | 'workflow-execution'
   | 'schedule-execution'
@@ -84,6 +91,14 @@ export interface Job<TPayload = unknown, TOutput = unknown> {
   status: JobStatus
   createdAt: Date
   startedAt?: Date
+  /**
+   * When the job reached its current status, required whenever that status is
+   * one of `TERMINAL_JOB_STATUSES`. Consumers derive both an end timestamp and
+   * an elapsed duration from it, so a terminal job that omits it reports null
+   * for each. A backend reading an eventually-consistent source must supply its
+   * best-known transition instant rather than leaving this unset — never the
+   * time of the read, which grows on every poll.
+   */
   completedAt?: Date
   attempts: number
   maxAttempts: number

apps/sim/lib/execution/cancel-workflow-execution.ts

@@ -12,7 +12,7 @@ import {
 } from '@/lib/execution/cancellation'
 import { createExecutionEventWriter, readExecutionMetaState } from '@/lib/execution/event-buffer'
 import { abortManualExecution } from '@/lib/execution/manual-cancellation'
-import { elapsedDurationMsSql } from '@/lib/logs/execution/duration'
+import { cancelledExecutionLogFields } from '@/lib/logs/execution/cancellation'
 import { captureServerEvent } from '@/lib/posthog/server'
 import {
   cancelWorkflowGroupExecution,
@@ -381,11 +381,7 @@ export async function cancelWorkflowExecution(
       const cancelledAt = new Date()
       await db
         .update(workflowExecutionLogs)
-        .set({
-          status: 'cancelled',
-          endedAt: cancelledAt,
-          totalDurationMs: elapsedDurationMsSql(cancelledAt),
-        })
+        .set(cancelledExecutionLogFields(cancelledAt))
         .where(
           and(
             eq(workflowExecutionLogs.executionId, executionId),

apps/sim/lib/logs/execution/cancellation.test.ts

@@ -0,0 +1,49 @@
+/**
+ * @vitest-environment node
+ */
+
+import { describe, expect, it, vi } from 'vitest'
+
+vi.unmock('drizzle-orm')
+vi.unmock('@sim/db')
+vi.unmock('@sim/db/schema')
+
+process.env.DATABASE_URL ??= 'postgresql://user:pass@localhost:5432/test'
+
+const { PgDialect } = await import('drizzle-orm/pg-core')
+const { cancelledExecutionLogFields } = await import('@/lib/logs/execution/cancellation')
+
+describe('cancelledExecutionLogFields', () => {
+  /**
+   * The five cancellation paths spread this payload into their own `.set()`.
+   * Hand-assembling it at each one had already dropped `executionDeadlineAt` at
+   * a single site, so the point of the factory is that the key set cannot vary
+   * between them — a field added here without a reason reaches all five.
+   */
+  it('writes exactly the terminal fields, deadline cleared', () => {
+    const endedAt = new Date('2026-08-13T12:00:05.000Z')
+
+    const fields = cancelledExecutionLogFields(endedAt)
+
+    expect(Object.keys(fields).sort()).toEqual([
+      'endedAt',
+      'executionDeadlineAt',
+      'status',
+      'totalDurationMs',
+    ])
+    expect(fields.status).toBe('cancelled')
+    expect(fields.endedAt).toBe(endedAt)
+    expect(fields.executionDeadlineAt).toBeNull()
+  })
+
+  /** `ended_at` and `total_duration_ms` must describe the same instant. */
+  it('derives the duration from the same instant it ends the run at', () => {
+    const endedAt = new Date('2026-08-13T12:00:05.000Z')
+
+    const { params } = new PgDialect().sqlToQuery(
+      cancelledExecutionLogFields(endedAt).totalDurationMs
+    )
+
+    expect(params).toContain(endedAt.toISOString())
+  })
+})

apps/sim/lib/logs/execution/cancellation.ts

@@ -0,0 +1,23 @@
+import { elapsedDurationMsSql } from '@/lib/logs/execution/duration'
+
+/**
+ * The fields every terminal cancellation sets on a `workflow_execution_logs`
+ * row, ready to spread into `.set()`.
+ *
+ * The five cancellation paths — direct, workflow-group with and without a
+ * sidecar, paused, and the async cancel route — differ in their database
+ * handle, their claim predicate, whether they read the row back, and what they
+ * do when the claim is lost, so they remain separate statements. What they must
+ * not differ in is the row they leave behind, and hand-assembling this payload
+ * at each one had already dropped `executionDeadlineAt` at a single site,
+ * leaving a cancelled run still carrying the deadline of an attempt that had
+ * stopped running.
+ */
+export function cancelledExecutionLogFields(endedAt: Date) {
+  return {
+    status: 'cancelled' as const,
+    endedAt,
+    totalDurationMs: elapsedDurationMsSql(endedAt),
+    executionDeadlineAt: null,
+  }
+}