fix(workflows): pin a stored block retry policy when loading it

`workflow_blocks.retry` is a jsonb column written verbatim. Three of its
writers never validate what they store: the realtime batch-add and
replace-state ops take untyped block records, and the admin/superuser
import routes persist externally-authored workflow JSON. `load.ts` then
asserted the blob was already a `BlockRetryConfig` and handed it straight
to the HTTP boundary, where `workflowBlockStateSchema` bounds `maxTries`
to 2..5 and `waitBetweenTriesMs` to 0..5000.

That schema is shared between the PUT `/state` body, where the bound is
right, and the GET `/api/workflows/[id]` and `/state` responses, where it
is fatal. The response `.parse` in the shared route builder throws a
ZodError, which is not an `OrchestrationError`, so the error policy
declines it and it falls through to a 500. One out-of-range or partial
stored value therefore made a workflow permanently unopenable, with no
in-product repair — every UI write path reads the workflow first.

The feature already declares clamp-on-read as its contract: the commit
that added it says bounds are clamped on read rather than rejected, the
TSDoc on `resolveBlockRetryConfig` says the same, and `block-retry.test.ts`
asserts it. Execution has always honoured that. Only the read boundary
disagreed, so that is what this fixes: the loader now constructs a real
`BlockRetryConfig` from the blob through `normalizeBlockRetryTries` /
`normalizeBlockRetryWaitMs`, filling defaults for missing fields and
carrying `enabled` across unchanged.

`loadWorkflowFromNormalizedTablesRaw` is the single read choke point for
both apps — `@sim/workflow-persistence` for the Next app and the realtime
server's full-state emit — so one edit repairs every reader, including
rows that are already out of range, and the row self-heals on the next
save. It matches `clampParallelBatchSize` a few lines below, which already
pins a stored subflow value on the same path.

Alternatives rejected:

- Validating on write. It leaves every existing bad row fatal forever, and
  it would have to be repeated across three realtime ops plus roughly a
  dozen `saveWorkflowToNormalizedTables` callers, none of which share a
  validation seam.
- Bounding `BlockRetrySchema` in `@sim/realtime-protocol`. Its own TSDoc is
  correct that batch-add and replace-state bypass it, so this closes one
  writer and leaves the 500.
- Relaxing the response contract. It stops the 500 but leaves the editor
  rendering a number execution will never run. A test now pins the write
  bound so that shortcut fails loudly.
- `resolveBlockRetryConfig`. It returns null for a disabled policy, which
  would erase the numbers a builder configured every time state is read.

Tests: six cases in `packages/workflow-persistence/src/load.test.ts` (four
red before this change) covering out-of-range enabled, out-of-range
disabled with `enabled` preserved, missing fields, a non-boolean flag, an
untouched in-range policy, and NULL meaning "runs once"; plus a contract
test that the write bound still rejects out-of-range input.

Author: waleedlatif1

apps/sim/lib/api/contracts/workflows.test.ts

@@ -6,6 +6,7 @@ import {
   internalCancelWorkflowExecutionReasonSchema,
   updateWorkflowBodySchema,
   workflowListItemSchema,
+  workflowStateSchema,
 } from '@/lib/api/contracts/workflows'
 import { PRIVATE_SECRET_PROVENANCE_FIELD } from '@/lib/execution/private-tool-metadata'
 
@@ -148,4 +149,45 @@ describe('workflow contracts', () => {
       expect(cancelWorkflowExecutionReasonSchema.options).not.toContain(reason)
     }
   })
+
+  /**
+   * `workflowStateSchema` is the PUT `/api/workflows/[id]/state` body and also
+   * the `state` slot of the GET response. A stored value outside these bounds
+   * used to 500 the read, which is now prevented by pinning the policy when the
+   * normalized tables are loaded — not by widening the write contract. Relaxing
+   * these bounds would let a caller persist a policy the executor will not run.
+   */
+  it('rejects a retry policy outside the bounds on the write contract', () => {
+    const stateWith = (retry: Record<string, unknown>) => ({
+      blocks: {
+        'block-1': {
+          id: 'block-1',
+          type: 'api',
+          name: 'API',
+          position: { x: 0, y: 0 },
+          subBlocks: {},
+          outputs: {},
+          enabled: true,
+          retry,
+        },
+      },
+      edges: [],
+    })
+
+    expect(
+      workflowStateSchema.safeParse(
+        stateWith({ enabled: true, maxTries: 999, waitBetweenTriesMs: 0 })
+      ).success
+    ).toBe(false)
+    expect(
+      workflowStateSchema.safeParse(
+        stateWith({ enabled: true, maxTries: 3, waitBetweenTriesMs: 10_000_000 })
+      ).success
+    ).toBe(false)
+    expect(
+      workflowStateSchema.safeParse(
+        stateWith({ enabled: true, maxTries: 3, waitBetweenTriesMs: 0 })
+      ).success
+    ).toBe(true)
+  })
 })

packages/workflow-persistence/src/load.test.ts

@@ -0,0 +1,154 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+type Row = Record<string, unknown>
+
+const tables = vi.hoisted(() => ({
+  workflow: { name: 'workflow' as const },
+  workflowBlocks: {
+    name: 'workflowBlocks' as const,
+    workflowId: 'workflow_id',
+    updatedAt: 'updated_at',
+  },
+  workflowEdges: { name: 'workflowEdges' as const, workflowId: 'workflow_id' },
+  workflowSubflows: { name: 'workflowSubflows' as const, workflowId: 'workflow_id' },
+}))
+
+vi.mock('@sim/db', () => ({
+  db: {},
+  workflow: tables.workflow,
+  workflowBlocks: tables.workflowBlocks,
+  workflowEdges: tables.workflowEdges,
+  workflowSubflows: tables.workflowSubflows,
+}))
+
+vi.mock('@sim/logger', () => ({
+  createLogger: () => ({ debug: vi.fn(), error: vi.fn(), info: vi.fn(), warn: vi.fn() }),
+}))
+
+vi.mock('drizzle-orm', () => ({
+  and: vi.fn(),
+  eq: vi.fn(),
+  getTableColumns: vi.fn(() => ({})),
+  isNull: vi.fn(),
+  sql: vi.fn(() => 'updated_at::text'),
+}))
+
+import { loadWorkflowFromNormalizedTablesRaw } from './load'
+
+/**
+ * Minimal stand-in for the drizzle query builder the loader uses: every chain
+ * ends in the rows registered for the table named by `.from(...)`, and the
+ * chain is awaitable both with and without a trailing `.limit(...)`.
+ */
+function createTx(rowsByTable: Record<string, Row[]>) {
+  const resultFor = (rows: Row[]) => {
+    const result = {
+      where: () => result,
+      limit: () => Promise.resolve(rows),
+      then: (onFulfilled: (rows: Row[]) => unknown) => Promise.resolve(rows).then(onFulfilled),
+    }
+    return result
+  }
+
+  return {
+    select: () => ({
+      from: (table: { name: string }) => resultFor(rowsByTable[table.name] ?? []),
+    }),
+  }
+}
+
+function blockRow(retry: unknown): Row {
+  return {
+    id: 'block-1',
+    type: 'api',
+    name: 'API',
+    positionX: '0',
+    positionY: '0',
+    enabled: true,
+    horizontalHandles: true,
+    advancedMode: false,
+    errorEnabled: false,
+    retry,
+    triggerMode: false,
+    height: '0',
+    subBlocks: {},
+    outputs: {},
+    data: {},
+    locked: false,
+    updatedAtText: '2026-01-01 00:00:00.000000',
+  }
+}
+
+async function loadRetry(retry: unknown) {
+  const tx = createTx({
+    workflowBlocks: [blockRow(retry)],
+    workflowEdges: [],
+    workflowSubflows: [],
+    workflow: [{ workspaceId: 'workspace-1' }],
+  })
+
+  const loaded = await loadWorkflowFromNormalizedTablesRaw(
+    'workflow-1',
+    tx as unknown as Parameters<typeof loadWorkflowFromNormalizedTablesRaw>[1]
+  )
+
+  return loaded?.blocks['block-1'].retry
+}
+
+describe('loadWorkflowFromNormalizedTablesRaw retry normalization', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+  })
+
+  /**
+   * The `retry` column is jsonb written verbatim by writers that never bound it
+   * (realtime batch-add and replace-state, the admin/superuser import routes),
+   * so a stored value can sit outside the range the HTTP read contract demands.
+   */
+  it('pins an out-of-range enabled policy to the bounds', async () => {
+    expect(
+      await loadRetry({ enabled: true, maxTries: 999, waitBetweenTriesMs: 10_000_000 })
+    ).toEqual({ enabled: true, maxTries: 5, waitBetweenTriesMs: 5000 })
+  })
+
+  /**
+   * A disabled policy keeps its configured numbers so switching retry off and
+   * back on restores them. This is what `resolveBlockRetryConfig` would destroy.
+   */
+  it('pins a disabled policy without discarding it', async () => {
+    expect(await loadRetry({ enabled: false, maxTries: 99, waitBetweenTriesMs: -4 })).toEqual({
+      enabled: false,
+      maxTries: 5,
+      waitBetweenTriesMs: 0,
+    })
+  })
+
+  it('fills the defaults for a policy stored with fields missing', async () => {
+    expect(await loadRetry({ enabled: true })).toEqual({
+      enabled: true,
+      maxTries: 3,
+      waitBetweenTriesMs: 1000,
+    })
+  })
+
+  it('resolves a non-boolean enabled flag the way execution reads it', async () => {
+    expect(await loadRetry({ enabled: 'yes', maxTries: 3, waitBetweenTriesMs: 1000 })).toEqual({
+      enabled: true,
+      maxTries: 3,
+      waitBetweenTriesMs: 1000,
+    })
+  })
+
+  it('leaves an in-range policy untouched', async () => {
+    expect(await loadRetry({ enabled: true, maxTries: 4, waitBetweenTriesMs: 250 })).toEqual({
+      enabled: true,
+      maxTries: 4,
+      waitBetweenTriesMs: 250,
+    })
+  })
+
+  /** NULL is reserved for a block that never had a policy: it runs once. */
+  it('reports no policy for a block that never had one', async () => {
+    expect(await loadRetry(null)).toBeUndefined()
+  })
+})

packages/workflow-persistence/src/load.ts

@@ -1,7 +1,9 @@
 import { db, workflow, workflowBlocks, workflowEdges, workflowSubflows } from '@sim/db'
 import { createLogger } from '@sim/logger'
-import type { BlockState, Loop, Parallel } from '@sim/workflow-types/workflow'
+import type { BlockRetryConfig, BlockState, Loop, Parallel } from '@sim/workflow-types/workflow'
 import {
+  normalizeBlockRetryTries,
+  normalizeBlockRetryWaitMs,
   normalizeWorkflowEdgeSourceHandle,
   normalizeWorkflowEdgeTargetHandle,
   SUBFLOW_TYPES,
@@ -13,6 +15,35 @@ import type { DbOrTx, NormalizedWorkflowData } from './types'
 
 const logger = createLogger('WorkflowPersistenceLoad')
 
+/**
+ * Rebuilds a stored retry policy as a real {@link BlockRetryConfig} instead of
+ * asserting that the raw `jsonb` blob already is one.
+ *
+ * The column is written verbatim by writers that never validate its contents —
+ * the realtime batch-add and replace-state ops take untyped block records, and
+ * the admin/superuser import routes persist externally-authored workflow JSON —
+ * so a row can hold an out-of-range number, a missing field, or a non-boolean
+ * flag. Pinning the values here is the policy the feature declares (see
+ * `resolveBlockRetryConfig`) and applies it to every reader at once: the editor
+ * renders exactly the numbers execution will use, and the strict HTTP contract
+ * that serves this state can never reject a workflow it is meant to open.
+ *
+ * `enabled` is carried across rather than resolved, so the numbers a builder
+ * configured survive switching retry off and back on. That is why
+ * `resolveBlockRetryConfig` — which collapses a disabled policy to `null` —
+ * must not be used on this path.
+ */
+function normalizeStoredBlockRetry(stored: unknown): BlockRetryConfig | undefined {
+  if (stored == null || typeof stored !== 'object') return undefined
+  const retry = stored as Record<string, unknown>
+
+  return {
+    enabled: Boolean(retry.enabled),
+    maxTries: normalizeBlockRetryTries(retry.maxTries),
+    waitBetweenTriesMs: normalizeBlockRetryWaitMs(retry.waitBetweenTriesMs),
+  }
+}
+
 export interface RawNormalizedWorkflow extends NormalizedWorkflowData {
   workspaceId: string
   /**
@@ -87,7 +118,7 @@ export async function loadWorkflowFromNormalizedTablesRaw(
         horizontalHandles: block.horizontalHandles,
         advancedMode: block.advancedMode,
         errorEnabled: block.errorEnabled,
-        retry: (block.retry as BlockState['retry']) ?? undefined,
+        retry: normalizeStoredBlockRetry(block.retry),
         triggerMode: block.triggerMode,
         height: Number(block.height),
         subBlocks: (block.subBlocks as BlockState['subBlocks']) || {},