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

* fix(execution): give a cancelled async run its terminal metadata The staging integration suite has been failing integ-cancel-async-api-key/async-execution-becomes-cancelled: the run reports status cancelled with a null endedAt and a null duration, so the assertion fails and the dependent worker-stop check never runs. The run resource falls back to the queue job whenever no execution-log row exists yet, and a cancel that lands before the worker has written that row leaves exactly that state. Trigger.dev marks a run canceled the moment it accepts the cancellation but only stamps its finish time when the worker drains, so for the seconds in between the job is terminal with no timestamp, and the projection faithfully reports a terminal status with nothing to date it. The run writes a correct log row when it finally drains, which is why the endpoint heals itself and the alarm fires intermittently rather than always. The backend was discarding the one timestamp that is always present: the retrieve response carries a required updatedAt beside the optional finishedAt. A finish time still wins wherever it exists, so nothing that already reports correctly changes, and the fallback is taken only once the mapped status is terminal — an active run's updatedAt marks progress, and reading it as an end would retire a run that is still going. It records the server's last transition for the run rather than the reader's clock, so it stays put across polls instead of growing. Also carries the duration on a fifth cancellation write, in the internal cancel route, that the earlier pass missed because its sweep covered lib and not app. * 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.
2026-09-24 15:45:35 +08:00 · 2026-08-13 20:07:37 -07:00
parent c6047efc4e
commit b9a70e43f9
16 changed files with 274 additions and 64 deletions
@@ -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()
@@ -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(
@@ -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,
    })
  })
@@ -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),
@@ -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', () => {
@@ -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,
@@ -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
@@ -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),
@@ -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())
+  })
+})
@@ -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,
+  }
+}
@@ -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)
  })
 })
@@ -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})`
 }
@@ -3,7 +3,7 @@ import { tableRowExecutions, userTableDefinitions, workflowExecutionLogs } from
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
 import { and, eq, inArray } from 'drizzle-orm'
-import { elapsedDurationMsSql } from '@/lib/logs/execution/duration'
+import { cancelledExecutionLogFields } from '@/lib/logs/execution/cancellation'
 import { appendTableEvent } from '@/lib/table/events'

 const logger = createLogger('WorkflowGroupCancellation')
@@ -153,12 +153,7 @@ export async function cancelWorkflowGroupExecution(
      const cancelledAt = new Date()
      const [cancelledLog] = await tx
        .update(workflowExecutionLogs)
-        .set({
-          status: 'cancelled',
-          endedAt: cancelledAt,
-          totalDurationMs: elapsedDurationMsSql(cancelledAt),
-          executionDeadlineAt: null,
-        })
+        .set(cancelledExecutionLogFields(cancelledAt))
        .where(
          and(
            eq(workflowExecutionLogs.workspaceId, options.workspaceId),
@@ -194,12 +189,7 @@ export async function cancelWorkflowGroupExecution(
    if (workflowLogActive) {
      const [cancelledLog] = await tx
        .update(workflowExecutionLogs)
-        .set({
-          status: 'cancelled',
-          endedAt: now,
-          totalDurationMs: elapsedDurationMsSql(now),
-          executionDeadlineAt: null,
-        })
+        .set(cancelledExecutionLogFields(now))
        .where(
          and(
            eq(workflowExecutionLogs.workspaceId, options.workspaceId),
@@ -128,6 +128,7 @@ describe('getWorkflowExecutionStatus queue projection', () => {
      status: 'cancelled',
      level: 'info',
      endedAt: '2026-08-05T12:00:01.000Z',
+      totalDurationMs: 1000,
      error: null,
    })
  })
@@ -33,7 +33,7 @@ import {
 } from '@/lib/execution/payloads/large-value-metadata'
 import { compactBlockLogs, compactExecutionPayload } from '@/lib/execution/payloads/serializer'
 import { preprocessExecution } from '@/lib/execution/preprocessing'
-import { elapsedDurationMsSql } from '@/lib/logs/execution/duration'
+import { cancelledExecutionLogFields } from '@/lib/logs/execution/cancellation'
 import { LoggingSession } from '@/lib/logs/execution/logging-session'
 import { cleanupExecutionBase64Cache } from '@/lib/uploads/utils/user-file-base64.server'
 import { executeWorkflowCore } from '@/lib/workflows/executor/execution-core'
@@ -2616,12 +2616,7 @@ export class PauseResumeManager {
      if (!cancellationAlreadyTerminal) {
        const [cancelledExecution] = await tx
          .update(workflowExecutionLogs)
-          .set({
-            status: 'cancelled',
-            endedAt: now,
-            totalDurationMs: elapsedDurationMsSql(now),
-            executionDeadlineAt: null,
-          })
+          .set(cancelledExecutionLogFields(now))
          .where(
            and(
              eq(workflowExecutionLogs.executionId, executionId),
@@ -440,6 +440,13 @@ export const workflowExecutionLogs = pgTable(
    /** Absolute deadline for the current active attempt; cleared while paused or terminal. */
    executionDeadlineAt: timestamp('execution_deadline_at'),
    endedAt: timestamp('ended_at'),
+    /**
+     * Wall clock from `started_at` for a terminal row; for a `pending` (paused)
+     * row, the active duration recorded at the checkpoint, which excludes the
+     * time the run sits waiting. Resuming leaves that checkpoint value in place
+     * while the row accrues time again, so a `running` row's value is stale
+     * until the next terminal write recomputes it.
+     */
    totalDurationMs: integer('total_duration_ms'),

    /**