refactor(execution): derive a cancelled run's terminal fields in one place

The five cancellation paths each hand-assembled the same four-key payload for
the workflow-execution log, and one of them had already drifted: the direct
cancel never cleared `execution_deadline_at`, leaving a cancelled row carrying
the deadline of an attempt that had stopped running. Extract the payload so the
key set cannot vary between them, and leave the paths themselves alone — they
differ in handle, claim predicate, whether they read the row back, and what they
do when the claim is lost, so they stay separate statements.

Bind the end instant through the `started_at` column encoder rather than a
pre-stringified ISO literal. `check:sql-date-binding` exists to enforce exactly
that binding; the literal passed only because it was already a string.

Collapse the duration expression to one `COALESCE` over a valueless-`ELSE`
`CASE`, which builds the elapsed fragment once instead of in both branches, and
reuse it for the stale-execution sweeper, which carried its own copy along with
a second int4 ceiling constant.

Document the invariants the fix depends on where a reader meets them: that
`total_duration_ms` means wall clock for a terminal row and active time for a
paused one, and that a terminal job must carry its transition instant.

Author: waleedlatif1

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.ts

@@ -24,7 +24,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 { workflowExecutionOriginSql } from '@/lib/logs/execution-origin'
 import { captureServerEvent } from '@/lib/posthog/server'
 import {
@@ -216,12 +216,7 @@ async function claimExecutionLogCancellation(args: {
   const now = new Date()
   const [cancelledExecution] = await db
     .update(workflowExecutionLogs)
-    .set({
-      status: 'cancelled',
-      endedAt: now,
-      totalDurationMs: elapsedDurationMsSql(now),
-      executionDeadlineAt: null,
-    })
+    .set(cancelledExecutionLogFields(now))
     .where(
       and(
         eq(workflowExecutionLogs.executionId, args.executionId),

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,12 +214,6 @@ function mapTriggerDevStatus(status: string): JobStatus {
   }
 }
 
-const TERMINAL_JOB_STATUSES: readonly JobStatus[] = [
-  JOB_STATUS.COMPLETED,
-  JOB_STATUS.FAILED,
-  JOB_STATUS.CANCELLED,
-]
-
 /**
  * Dates the end of a run that trigger.dev already reports as terminal.
  *

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,
+  }
+}

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

@@ -29,14 +29,16 @@ describe('elapsedDurationMsSql', () => {
   })
 
   /**
-   * `started_at` is `timestamp without time zone` holding a UTC wall clock. A
-   * driver-bound `Date` infers `timestamptz`, which would make the interval
-   * depend on the session zone; the explicit cast is what keeps it stable.
+   * `started_at` is `timestamp without time zone` holding a UTC wall clock. The
+   * end instant is bound through that column's mapper, which emits a UTC ISO
+   * string; the explicit cast drops the offset to the same naive reading rather
+   * than leaving the interval to depend on how Postgres resolves the operator.
    */
   it('binds the end instant as a zone-free timestamp', () => {
     const { sql, params } = render(new Date('2026-08-13T12:00:05.000Z'))
 
-    expect(sql).toContain('::timestamp')
+    expect(sql).toContain('::timestamp -')
+    expect(sql).not.toContain('::timestamptz')
     expect(params).toContain('2026-08-13T12:00:05.000Z')
     expect(params.some((param) => Array.isArray(param))).toBe(false)
   })
@@ -71,8 +73,7 @@ describe('elapsedDurationMsSql', () => {
   it('keeps the duration a paused run already recorded', () => {
     const { sql } = render(new Date('2026-08-13T12:00:05.000Z'))
 
-    expect(sql).toContain(`"status" = 'pending' THEN COALESCE(`)
-    expect(sql).toContain('"total_duration_ms"')
+    expect(sql).toContain(`"status" = 'pending' THEN "workflow_execution_logs"."total_duration_ms"`)
   })
 
   /**
@@ -83,9 +84,23 @@ describe('elapsedDurationMsSql', () => {
   it('recomputes for a running row rather than trusting a stale checkpoint', () => {
     const { sql } = render(new Date('2026-08-13T12:00:05.000Z'))
 
-    const elseBranch = sql.slice(sql.indexOf('ELSE'))
-    expect(elseBranch).toContain('LEAST(')
-    expect(elseBranch).not.toContain('COALESCE(')
-    expect(elseBranch).not.toContain('"total_duration_ms"')
+    const fallback = sql.slice(sql.indexOf('END,'))
+    expect(fallback).toContain('LEAST(')
+    expect(fallback).not.toContain('"total_duration_ms"')
+  })
+
+  /**
+   * The preserved-checkpoint rule and the elapsed fallback are one `COALESCE`
+   * over a valueless-`ELSE` `CASE`, not a `CASE` repeating the elapsed
+   * expression in both branches. `status` is `NOT NULL` and `started_at` is
+   * `NOT NULL`, so the `CASE` yields NULL exactly for a non-`pending` row or a
+   * `pending` row that never recorded one — the two cases that must fall
+   * through — and the fallback can never itself be NULL.
+   */
+  it('builds the elapsed expression once', () => {
+    const { sql, params } = render(new Date('2026-08-13T12:00:05.000Z'))
+
+    expect(sql.match(/LEAST\(/g)).toHaveLength(1)
+    expect(params).toHaveLength(2)
   })
 })

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

@@ -1,6 +1,8 @@
 import { workflowExecutionLogs } from '@sim/db/schema'
 import { type SQL, sql } from 'drizzle-orm'
 
+const INT4_MAX_MS = 2_147_483_647
+
 /**
  * Elapsed run time for a terminal write that does not go through
  * `completeWorkflowExecution`, expressed against the row's own `started_at`.
@@ -12,13 +14,19 @@ import { type SQL, sql } from 'drizzle-orm'
  * `GET /api/v2/logs` — a null there reads as "no duration recorded" and drops
  * the run out of every `minDurationMs`/`maxDurationMs` query.
  *
- * `ended_at` is bound as an explicit `timestamp` rather than a `Date`, because
- * `started_at` is `timestamp without time zone` holding a UTC wall clock: an
- * ISO string casts to the same naive reading, while a driver-bound `Date`
- * would infer `timestamptz` and make the interval depend on the session zone.
+ * `ended_at` is bound through `started_at`'s own column mapper and cast to
+ * `timestamp`, because that column is `timestamp without time zone` holding a
+ * UTC wall clock. The cast is what pins the reading: the mapper emits a UTC ISO
+ * instant, and casting it drops the offset to the same naive wall clock the
+ * column stores. Left uncast the interval would depend on Postgres resolving
+ * the subtraction to the `timestamp` overload rather than `timestamptz`, and
+ * getting that wrong yields a duration in the wrong zone rather than an error.
  *
- * Floored at 1ms to match `completeWorkflowExecution`, so a cancellation that
- * lands inside the same millisecond as the start still records that it ran.
+ * Floored at 1ms to match the `Math.max(1, durationMs)` the completion path
+ * applies, so a cancellation landing inside the same millisecond as the start
+ * still records that it ran. The floor doubles as the guard against a negative
+ * interval, which is otherwise reachable whenever the clock that stamped
+ * `started_at` runs ahead of the one cancelling.
  *
  * Saturated at the column's own ceiling rather than left to overflow. The
  * column is `integer`, so an untimed run cancelled after ~24.8 days would
@@ -41,9 +49,8 @@ import { type SQL, sql } from 'drizzle-orm'
  * measures wall clock. A `running` row therefore always recomputes; only a
  * `pending` one — paused, and not accruing — keeps what it has.
  */
-const INT4_MAX_MS = 2_147_483_647
-
 export function elapsedDurationMsSql(endedAt: Date): SQL<number> {
-  const elapsed = sql`LEAST(${INT4_MAX_MS}, GREATEST(1, ROUND(EXTRACT(EPOCH FROM (${endedAt.toISOString()}::timestamp - ${workflowExecutionLogs.startedAt})) * 1000)))::integer`
-  return sql<number>`CASE WHEN ${workflowExecutionLogs.status} = 'pending' THEN COALESCE(${workflowExecutionLogs.totalDurationMs}, ${elapsed}) ELSE ${elapsed} END`
+  const endedAtParam = sql.param(endedAt, workflowExecutionLogs.startedAt)
+  const elapsed = sql`LEAST(${INT4_MAX_MS}, GREATEST(1, ROUND(EXTRACT(EPOCH FROM (${endedAtParam}::timestamp - ${workflowExecutionLogs.startedAt})) * 1000)))::integer`
+  return sql<number>`COALESCE(CASE WHEN ${workflowExecutionLogs.status} = 'pending' THEN ${workflowExecutionLogs.totalDurationMs} END, ${elapsed})`
 }