fix(schedule): reconcile interrupted schedule executions (#6780)

* fix schedule execution recovery

* simplify schedule recovery provider handling

* fix(schedules): stop reconciliation rewriting healthy carriers and index its scan

Follow-up hardening on the schedule recovery reconciliation.

Reconciliation settled every carrier it examined, including ones that had
already reached a terminal status and recorded their own outcome. Because the
normal completion path never stamps the reconciled marker, each successful
run's carrier was picked up on the next tick and rewritten: `completedAt`
bumped, `error` nulled and `output` replaced with a recovery stub, all of which
`GET /api/jobs/[jobId]` surfaces. A completed carrier whose execution log had
aged out was additionally flipped to failed. Settle only carriers still in
flight; a terminal one is owed schedule accounting and the marker, nothing more.

The recovery scan matched no index. There was none on `async_jobs.updated_at`,
and the unreconciled-terminal branch tested a jsonb extraction, so the whole
OR fell back to a sequential scan and sort of `async_jobs` on every tick. Add
the partial index, and spell the branch's status list and metadata key as SQL
literals: Postgres cannot prove a parameterised predicate implies a literal
index predicate, so a bound key would have left the new index unused.

Irrecoverable carrier tombstones were exempted from retention with no secondary
expiry, so the one class of row that can never reconcile grew without bound.
Give them a longer bounded window instead.

`WORKFLOW_EXECUTION_MAX_ROWS_PER_RUN` had become a per-status budget when the
stale-execution sweep gained its `redacting` pass, silently doubling the
per-run cap. Share the budget across both passes, and add the matching partial
indexes so the second pass keeps an index rather than seq-scanning.

Also consolidates the carrier metadata keys, their predicates and the jsonb
merge into one module -- the two routes were the writer and reader of the same
keys with no shared symbol, so a rename type-checked clean while silently
breaking retention.

* fix(schedules): rotate deferred carriers and stop the redacting sweep failing live runs

Addresses the review findings on the reconciliation hardening.

A carrier whose accounting was deferred got no write at all, so it kept its
`updatedAt` and stayed at the head of the `updatedAt`-ordered recovery batch on
every tick, starving every other claimed carrier and never becoming eligible for
retention. This was a regression from the previous commit: settling used to bump
the timestamp for every examined row, and skipping the settle for already-terminal
carriers removed the bump with it. A payload with no `scheduledFor` can never
reconcile, so such a row pinned a batch slot permanently. Bump `updatedAt`
unconditionally and keep only the reconciled marker conditional.

The stale-execution sweep terminalized `redacting` logs on the execution deadline.
That deadline bounds execution, while `redacting` covers payload masking after the
run already finished -- so a run that used most of its budget entered redaction
with the deadline due, and the sweep failed it five minutes later while the worker
was still masking. Schedule recovery then read the log as a failed occurrence and
counted a failure that never happened, even though the worker's terminal write
later restored `completed`. Sweep `redacting` on the generic stale window only.

`getScheduleNextRunAt` falls back to a daily cadence when a schedule has no cron
expression. Deployment cannot persist such a schedule, so the branch is
unreachable, but this change widened its use from failure recovery to every
outcome -- log a warning when it fires rather than silently guessing a cadence.

`executionDeadlineAt` was missing from the shared `workflowExecutionLogs` schema
mock, so it read as `undefined` and assertions comparing against that column were
trivially true. Add it.

* refactor(schedules): drop vestigial recovery code and close a builder gap

Follow-ups from a full re-read of the change. No behavior change except the
removed dead code paths.

The metadata merge stripped a `scheduleRecoveryBlocked` key on every write. That
key has never been written by any shipped code -- it appears nowhere in staging
and nowhere in history outside this branch -- so the strip guarded against a
state that cannot exist, at the cost of an extra jsonb operation and a bind
parameter on every reconciliation write.

`processScheduleItem` set `carrierObservedOrLookupUncertain` immediately before
returning on an ambiguous enqueue. The flag is only read from the surrounding
catch block, which a normal return skips, so the assignment was dead and read as
though it were load-bearing. Replaced with a comment stating why the occurrence
is preserved.

The stale-execution sweep carried two near-identical `jsonb_set` templates that
differed only in their error expression, kept flat because the test mock renders
nested SQL fragments as placeholders. The suite now has a recursive renderer, so
the error expression is a named per-status value and there is one `jsonb_set`.

Success was the only schedule outcome without a named update builder, which left
`executeScheduleJob` using two idioms for the same guarded write and left the
update shape untested. Add `buildScheduleSuccessUpdate` beside its cancellation
and failure siblings, use it from both call sites, and cover it the way
`buildScheduleCancellationUpdate` is covered -- a mutation of its `failedCount`
reset previously passed every suite.

---------

Co-authored-by: Bill Leoutsakos <billleoutsakos@Bills-MacBook-Pro.local>
Co-authored-by: Waleed Latif <walif6@gmail.com>

Author: 3 people

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

@@ -5,7 +5,16 @@ import { asyncJobs, tableJobs, workflowExecutionLogs } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { createMockRequest, dbChainMockFns, queueTableRows, resetDbChainMock } from '@sim/testing'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
-import { MAX_JOB_DURATION_SECONDS, MIN_JOB_DURATION_SECONDS } from '@/lib/core/async-jobs'
+import {
+  JOB_RETENTION_HOURS,
+  MAX_JOB_DURATION_SECONDS,
+  MIN_JOB_DURATION_SECONDS,
+} from '@/lib/core/async-jobs'
+import {
+  SCHEDULE_CARRIER_IRRECOVERABLE_METADATA_KEY,
+  SCHEDULE_CARRIER_IRRECOVERABLE_RETENTION_HOURS,
+  SCHEDULE_CARRIER_RECONCILED_METADATA_KEY,
+} from '@/lib/workflows/schedules/carrier-metadata'
 
 const { mockDeleteFile, mockVerifyCronAuth } = vi.hoisted(() => ({
   mockDeleteFile: vi.fn().mockResolvedValue(undefined),
@@ -27,6 +36,7 @@ interface MockCondition {
   conditions?: unknown[]
   left?: unknown
   right?: unknown
+  column?: unknown
   values?: unknown
   toSQL?: () => { sql: string; params: unknown[] }
 }
@@ -61,6 +71,33 @@ function createRequest() {
   )
 }
 
+/** Recursively renders a mocked drizzle fragment, expanding nested fragments. */
+function renderSql(fragment: unknown): string {
+  if (fragment === null || fragment === undefined) return ''
+  if (typeof fragment !== 'object') return String(fragment)
+  const candidate = fragment as { rawSql?: string; strings?: string[]; values?: unknown[] }
+  if (typeof candidate.rawSql === 'string') return candidate.rawSql
+  if (!candidate.strings) return ''
+  return candidate.strings
+    .map((part, index) =>
+      index < (candidate.values?.length ?? 0)
+        ? `${part}${renderSql(candidate.values?.[index])}`
+        : part
+    )
+    .join('')
+}
+
+/** Recursively collects the non-fragment bind values of a mocked fragment. */
+function collectSqlParams(fragment: unknown): unknown[] {
+  if (!fragment || typeof fragment !== 'object') return []
+  const candidate = fragment as { values?: unknown[] }
+  if (!candidate.values) return []
+  return candidate.values.flatMap((value) => {
+    const nested = collectSqlParams(value)
+    return nested.length > 0 ? nested : [value]
+  })
+}
+
 describe('stale execution cleanup deadline grace', () => {
   beforeEach(() => {
     vi.clearAllMocks()
@@ -103,25 +140,18 @@ describe('stale execution cleanup deadline grace', () => {
         totalDurationMs: { toSQL: () => { sql: string; params: unknown[] } }
         executionData: { toSQL: () => { sql: string; params: unknown[] } }
       }
-      const errorExpression = update.executionData.toSQL()
-      const staleDurationExpression = errorExpression.params.find(
-        (value): value is { toSQL: () => { sql: string; params: unknown[] } } =>
-          typeof value === 'object' &&
-          value !== null &&
-          'toSQL' in value &&
-          value.toSQL().sql.includes('EXTRACT(EPOCH')
-      )
       const totalDurationLeaves = flattenSqlParams(update.totalDurationMs.toSQL())
-
-      expect(errorExpression.sql).toContain('CASE')
-      expect(errorExpression.sql).toContain('IS NOT NULL')
-      expect(errorExpression.params).toContain(workflowExecutionLogs.executionDeadlineAt)
-      expect(errorExpression.params).toContain('Execution timed out')
-      expect(errorExpression.params).toContain(
-        'Execution terminated: worker timeout or crash after '
-      )
-      expect(staleDurationExpression?.toSQL().sql).toContain('ROUND')
-      expect(staleDurationExpression?.toSQL().params).toContain(workflowExecutionLogs.startedAt)
+      const renderedError = renderSql(update.executionData)
+      const errorLeaves = collectSqlParams(update.executionData)
+
+      expect(renderedError).toContain('CASE')
+      expect(renderedError).toContain('IS NOT NULL')
+      expect(renderedError).toContain('ROUND')
+      expect(renderedError).toContain('EXTRACT(EPOCH')
+      expect(errorLeaves).toContain(workflowExecutionLogs.executionDeadlineAt)
+      expect(errorLeaves).toContain('Execution timed out')
+      expect(errorLeaves).toContain('Execution terminated: worker timeout or crash after ')
+      expect(errorLeaves).toContain(workflowExecutionLogs.startedAt)
       expect(totalDurationLeaves).toContain(2_147_483_647)
       expect(totalDurationLeaves).toContain(workflowExecutionLogs.startedAt)
       expect(totalDurationLeaves).toContainEqual(new Date('2026-08-03T12:10:00.000Z'))
@@ -131,6 +161,50 @@ describe('stale execution cleanup deadline grace', () => {
     }
   })
 
+  it('sweeps redacting logs on the generic window, never the execution deadline', async () => {
+    const response = await GET(createRequest())
+
+    expect(response.status).toBe(200)
+    const redactingPredicates = dbChainMockFns.where.mock.calls
+      .map(([condition]) => flattenConditions(condition))
+      .filter((conditions) =>
+        conditions.some(
+          (condition) =>
+            condition.type === 'eq' &&
+            condition.left === workflowExecutionLogs.status &&
+            condition.right === 'redacting'
+        )
+      )
+
+    expect(redactingPredicates.length).toBeGreaterThan(0)
+    for (const conditions of redactingPredicates) {
+      expect(
+        conditions.some((condition) => condition.left === workflowExecutionLogs.executionDeadlineAt)
+      ).toBe(false)
+      expect(
+        conditions.some(
+          (condition) =>
+            condition.type === 'lt' && condition.left === workflowExecutionLogs.startedAt
+        )
+      ).toBe(true)
+    }
+  })
+
+  it('terminalizes stale running and redacting execution logs', async () => {
+    const response = await GET(createRequest())
+
+    expect(response.status).toBe(200)
+    const statusPredicates = dbChainMockFns.where.mock.calls
+      .flatMap(([condition]) => flattenConditions(condition))
+      .filter(
+        (condition) => condition.type === 'eq' && condition.left === workflowExecutionLogs.status
+      )
+
+    expect(statusPredicates.map(({ right }) => right)).toEqual(
+      expect.arrayContaining(['running', 'redacting'])
+    )
+  })
+
   it('reports a worker cleanup deadline while preserving the generic stale fallback', async () => {
     queueTableRows(asyncJobs, [{ id: 'async-job-1' }])
     dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'async-job-1' }])
@@ -175,6 +249,108 @@ describe('stale execution cleanup deadline grace', () => {
     )
   })
 
+  it('leaves pending and processing schedule jobs to schedule recovery', async () => {
+    const response = await GET(createRequest())
+
+    expect(response.status).toBe(200)
+    const activeAsyncPredicates = dbChainMockFns.where.mock.calls
+      .map(([condition]) => flattenConditions(condition))
+      .filter((conditions) =>
+        conditions.some(
+          (condition) =>
+            condition.type === 'ne' &&
+            condition.left === asyncJobs.type &&
+            condition.right === 'schedule-execution'
+        )
+      )
+
+    expect(
+      activeAsyncPredicates.some((conditions) =>
+        conditions.some(
+          (condition) =>
+            condition.type === 'eq' &&
+            condition.left === asyncJobs.status &&
+            condition.right === 'processing'
+        )
+      )
+    ).toBe(true)
+    expect(
+      activeAsyncPredicates.some((conditions) =>
+        conditions.some(
+          (condition) =>
+            condition.type === 'eq' &&
+            condition.left === asyncJobs.status &&
+            condition.right === 'pending'
+        )
+      )
+    ).toBe(true)
+  })
+
+  it('retains terminal schedule carriers until reconciliation is recorded', async () => {
+    const response = await GET(createRequest())
+
+    expect(response.status).toBe(200)
+    const retentionConditions = dbChainMockFns.where.mock.calls.flatMap(([condition]) =>
+      flattenConditions(condition)
+    )
+    const reconciliationMarker = retentionConditions.find((condition) =>
+      renderSql(condition).includes(SCHEDULE_CARRIER_RECONCILED_METADATA_KEY)
+    )
+
+    expect(collectSqlParams(reconciliationMarker)).toContain(asyncJobs.metadata)
+    expect(
+      retentionConditions.some(
+        (condition) =>
+          condition.type === 'ne' &&
+          condition.left === asyncJobs.type &&
+          condition.right === 'schedule-execution'
+      )
+    ).toBe(true)
+  })
+
+  it('spells carrier metadata keys as SQL literals so the partial index matches', async () => {
+    const response = await GET(createRequest())
+
+    expect(response.status).toBe(200)
+    const reconciliationMarker = dbChainMockFns.where.mock.calls
+      .flatMap(([condition]) => flattenConditions(condition))
+      .find((condition) => renderSql(condition).includes(SCHEDULE_CARRIER_RECONCILED_METADATA_KEY))
+
+    expect(renderSql(reconciliationMarker)).toContain(
+      `'${SCHEDULE_CARRIER_RECONCILED_METADATA_KEY}'`
+    )
+    expect(collectSqlParams(reconciliationMarker)).not.toContain(
+      SCHEDULE_CARRIER_RECONCILED_METADATA_KEY
+    )
+  })
+
+  it('deletes irrecoverable schedule carrier tombstones once their longer window lapses', async () => {
+    const response = await GET(createRequest())
+
+    expect(response.status).toBe(200)
+    const retentionConditions = dbChainMockFns.where.mock.calls.flatMap(([condition]) =>
+      flattenConditions(condition)
+    )
+    const irrecoverableExclusion = retentionConditions.find((condition) =>
+      renderSql(condition).includes(SCHEDULE_CARRIER_IRRECOVERABLE_METADATA_KEY)
+    )
+
+    expect(renderSql(irrecoverableExclusion)).toContain("<> 'true'")
+    expect(collectSqlParams(irrecoverableExclusion)).toContain(asyncJobs.metadata)
+
+    const tombstoneWindow = retentionConditions.filter(
+      (condition) =>
+        condition.type === 'lt' &&
+        condition.left === asyncJobs.completedAt &&
+        condition.right instanceof Date
+    )
+    const oldest = Math.min(...tombstoneWindow.map(({ right }) => (right as Date).getTime()))
+    const newest = Math.max(...tombstoneWindow.map(({ right }) => (right as Date).getTime()))
+    expect(newest - oldest).toBe(
+      (SCHEDULE_CARRIER_IRRECOVERABLE_RETENTION_HOURS - JOB_RETENTION_HOURS) * 60 * 60 * 1000
+    )
+  })
+
   it('keeps table-job heartbeat cleanup independent from workflow timeout policy', async () => {
     vi.useFakeTimers()
     vi.setSystemTime(new Date('2026-08-03T12:00:00.000Z'))
@@ -220,8 +396,8 @@ describe('stale execution cleanup deadline grace', () => {
     const response = await GET(createRequest())
 
     expect(response.status).toBe(200)
-    expect(dbChainMockFns.transaction).toHaveBeenCalledTimes(7)
-    expect(dbChainMockFns.for).toHaveBeenCalledTimes(7)
+    expect(dbChainMockFns.transaction).toHaveBeenCalledTimes(8)
+    expect(dbChainMockFns.for).toHaveBeenCalledTimes(8)
     for (const [strength, options] of dbChainMockFns.for.mock.calls) {
       expect(strength).toBe('update')
       expect(options).toEqual({ skipLocked: true })