Function writeWorkflowGroupState

apps/sim/lib/table/cell-write.ts:41–141

(
  ctx: WriteWorkflowGroupContext,
  payload: WriteWorkflowGroupStatePayload
)

 * short-circuit any follow-up writes / job dispatch.
 */
export async function writeWorkflowGroupState(
  ctx: WriteWorkflowGroupContext,
  payload: WriteWorkflowGroupStatePayload
): Promise<'wrote' | 'skipped'> {
  const { tableId, rowId, workspaceId, groupId, executionId } = ctx
  const requestId = ctx.requestId ?? `wfgrp-${executionId}`
  const { getTableById } = await import('@/lib/table/service')
  const { getRowById, updateRow } = await import('@/lib/table/rows/service')

  const table = await getTableById(tableId)
  if (!table) {
    logger.warn(`Table ${tableId} vanished before group state write`)
    return 'wrote'
  }
  const row = await getRowById(tableId, rowId, workspaceId)
  if (!row) {
    logger.warn(`Row ${rowId} vanished before group state write`)
    return 'wrote'
  }
  const current = row.executions?.[groupId] as RowExecutionMetadata | undefined
  // Stale-worker guard: only blocks writes FROM an old worker (status =
  // running / completed / error / pending). A `queued` stamp from the
  // scheduler can claim the cell for a brand-new run — that's the new
  // authority. Same for `cancelled` (always authoritative, written by stop).
  const isCancelStamp = payload.executionState.status === 'cancelled'
  const isQueuedStamp = payload.executionState.status === 'queued'
  const isNewQueuedStamp = isQueuedStamp && current?.executionId !== executionId
  const bypassStaleWorker = isNewQueuedStamp || isCancelStamp
  if (!bypassStaleWorker && current && current.executionId && current.executionId !== executionId) {
    logger.info(
      `Skipping group write — stale worker (table=${tableId} row=${rowId} group=${groupId} mine=${executionId} active=${current.executionId})`
    )
    return 'skipped'
  }
  // A late `queued` stamp for the SAME run that's already moved past queued
  // (worker called markWorkflowGroupPickedUp before our parallel stamp landed)
  // must NOT overwrite the further-along state. Without this, a cell can show
  // "queued" forever while the worker is actually running.
  if (isQueuedStamp && current?.executionId === executionId && current.status !== 'pending') {
    logger.info(
      `Skipping queued stamp — same run already at status=${current.status} (table=${tableId} row=${rowId} group=${groupId} executionId=${executionId})`
    )
    return 'skipped'
  }
  // A `cancelled` cell rejects any worker write regardless of executionId — a
  // stop click can only stamp the dispatcher pre-stamp's executionId (often
  // null), so an executionId-matched guard would let the worker that later
  // claims the cell with its real id resurrect it. `bypassStaleWorker` (a fresh
  // `queued` claim from a new dispatch, or the authoritative cancel write
  // itself) still passes; manual re-runs clear the tombstone before stamping.
  if (!bypassStaleWorker && isExecCancelled(current)) {
    logger.info(
      `Skipping group write — cancelled (table=${tableId} row=${rowId} group=${groupId} executionId=${executionId})`
    )
    return 'skipped'
  }
  // Skip writing `cancelled` state with the guard — that's an authoritative
  // write from `cancelWorkflowGroupRuns` and must always land. New `queued`