fix(table): run a drained pre-stamp under the subject that stamped it

A dispatcher pre-stamp outlives the worker that wrote it: a cell task that
finds the row's cascade lock held bails, and the lock owner drains the
marker under its own payload. With two dispatches overlapping on one row
that ran another dispatch's request against the wrong person's tool
denylist — or, when the owner was an actorless auto-fire, against none.
The marker now carries its governed subject and the drain reads it back;
0316 also adds the partial index account deletion's cancel filter needs
and repeats 0315's backfill to close its rolling-deploy window.

Author: waleedlatif1

apps/sim/background/drain-governed-subject.test.ts

@@ -0,0 +1,152 @@
+/**
+ * @vitest-environment node
+ */
+import { resetDbChainMock } from '@sim/testing'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const mocks = vi.hoisted(() => ({
+  getTableById: vi.fn(),
+  getRowById: vi.fn(),
+  pickNextEligibleGroupForRow: vi.fn(),
+  writeWorkflowGroupState: vi.fn(),
+  markWorkflowGroupPickedUp: vi.fn(),
+  runEnrichment: vi.fn(),
+  getEnrichment: vi.fn(),
+  readStampedCapabilitySubject: vi.fn(),
+  checkAttributedUsageLimits: vi.fn(),
+  loadTableRowSecretProvenance: vi.fn(),
+}))
+
+vi.mock('@/lib/table/service', () => ({ getTableById: mocks.getTableById }))
+vi.mock('@/lib/table/rows/service', () => ({ getRowById: mocks.getRowById, updateRow: vi.fn() }))
+vi.mock('@/lib/table/rows/executions', () => ({
+  readStampedCapabilitySubject: mocks.readStampedCapabilitySubject,
+}))
+vi.mock('@/lib/table/workflow-columns', () => ({
+  pickNextEligibleGroupForRow: mocks.pickNextEligibleGroupForRow,
+  stashCellContextForResume: vi.fn(),
+  buildWorkflowGroupExecutionCorrelation: vi.fn(),
+}))
+vi.mock('@/lib/table/cell-write', () => ({
+  buildCancelledExecution: vi.fn(),
+  createWorkflowCellProgressWriter: vi.fn(),
+  writeWorkflowGroupState: mocks.writeWorkflowGroupState,
+  markWorkflowGroupPickedUp: mocks.markWorkflowGroupPickedUp,
+}))
+vi.mock('@/lib/table/workflow-cell-result', () => ({
+  classifyWorkflowCellTerminalResult: vi.fn(),
+}))
+vi.mock('@/enrichments/registry', () => ({ getEnrichment: mocks.getEnrichment }))
+vi.mock('@/enrichments/run', () => ({
+  runEnrichment: mocks.runEnrichment,
+  skippedEnrichmentDetail: () => ({}),
+}))
+vi.mock('@/lib/billing/core/billing-attribution', () => ({
+  assertBillingAttributionSnapshot: (snapshot: unknown) => snapshot,
+  checkAttributedUsageLimits: mocks.checkAttributedUsageLimits,
+  toBillingContext: () => ({}),
+}))
+vi.mock('@/lib/table/rows/secret-provenance', () => ({
+  createExactEmptyTableRowSecretProvenance: () => ({ complete: true, columns: {} }),
+  createTableRowSecretProvenanceFromRegistry: () => ({ complete: true, columns: {} }),
+  loadTableRowSecretProvenance: mocks.loadTableRowSecretProvenance,
+}))
+vi.mock('@/lib/table/events', () => ({ appendTableEvent: vi.fn() }))
+vi.mock('@/lib/table/dispatcher', () => ({
+  readDispatch: vi.fn(async () => ({ id: 'tdsp_carrier', status: 'dispatching' })),
+  completeDispatchIfActive: vi.fn(),
+}))
+
+import { runRowCascadeLoop } from '@/background/workflow-column-execution'
+
+function enrichmentGroup(id: string) {
+  return {
+    id,
+    type: 'enrichment',
+    enrichmentId: 'enrich-1',
+    workflowId: '',
+    outputs: [{ blockId: '', path: 'out', columnName: 'out' }],
+    inputMappings: [{ inputName: 'domain', columnName: 'domain' }],
+  }
+}
+
+const TABLE = {
+  id: 'table-1',
+  name: 'Table',
+  workspaceId: 'workspace-1',
+  schema: {
+    columns: [{ id: 'domain', name: 'domain', type: 'string' }],
+    workflowGroups: [enrichmentGroup('group-1'), enrichmentGroup('group-2')],
+  },
+}
+
+/** The carrier belongs to an actorless auto-fire: no subject, so no tool gate. */
+const CARRIER = {
+  tableId: 'table-1',
+  tableName: 'Table',
+  rowId: 'row-1',
+  groupId: 'group-1',
+  workflowId: '',
+  enrichmentId: 'enrich-1',
+  workspaceId: 'workspace-1',
+  executionId: 'execution-1',
+  dispatchId: 'tdsp_carrier',
+  executionTimeoutMs: 10_000,
+  capabilityGovernedUserId: null,
+  billingAttribution: {
+    actorUserId: 'user-1',
+    workspaceId: 'workspace-1',
+    organizationId: null,
+    billedAccountUserId: 'user-1',
+    billingEntity: { type: 'user' as const, id: 'user-1' },
+    billingPeriod: { start: '2026-07-01T00:00:00.000Z', end: '2026-08-01T00:00:00.000Z' },
+    payerSubscription: null,
+  },
+} as Parameters<typeof runRowCascadeLoop>[0]
+
+describe('draining another dispatch’s pre-stamped marker', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetDbChainMock()
+    mocks.getTableById.mockResolvedValue(TABLE)
+    mocks.getEnrichment.mockReturnValue({
+      id: 'enrich-1',
+      name: 'Enrich',
+      inputs: [{ id: 'domain', required: true }],
+      outputs: [{ id: 'out' }],
+    })
+    mocks.checkAttributedUsageLimits.mockResolvedValue({ isExceeded: false })
+    mocks.markWorkflowGroupPickedUp.mockResolvedValue('picked-up')
+    mocks.writeWorkflowGroupState.mockResolvedValue('wrote')
+    mocks.loadTableRowSecretProvenance.mockResolvedValue({
+      scope: { userId: 'user-1', workspaceId: 'workspace-1' },
+      byRowId: {},
+    })
+    mocks.runEnrichment.mockResolvedValue({ result: { out: 'x' }, cost: 0, detail: {} })
+    mocks.readStampedCapabilitySubject.mockResolvedValue('requesting-member')
+    // group-1 completes, then group-2 is picked up carrying an unclaimed marker.
+    mocks.getRowById.mockResolvedValue({
+      id: 'row-1',
+      data: { domain: 'example.com' },
+      executions: { 'group-2': { status: 'pending', executionId: null, workflowId: '' } },
+    })
+    mocks.pickNextEligibleGroupForRow
+      .mockReturnValueOnce(enrichmentGroup('group-2'))
+      .mockReturnValue(null)
+  })
+
+  /**
+   * The lock owner drains markers it did not stamp. Running them under its own
+   * subject applies the wrong person's tool denylist — and when the owner is an
+   * actorless auto-fire, no denylist at all.
+   */
+  it('runs the drained cell under the subject stamped with it', async () => {
+    await runRowCascadeLoop(CARRIER)
+
+    expect(mocks.readStampedCapabilitySubject).toHaveBeenCalledWith('row-1', 'group-2')
+    const subjects = mocks.runEnrichment.mock.calls.map(
+      ([, , ctx]) => (ctx as { userId: string | null }).userId
+    )
+    expect(subjects).toEqual([null, 'requesting-member'])
+  })
+})

apps/sim/background/workflow-column-execution.ts

@@ -301,6 +301,7 @@ export async function executeWorkflowGroupCellJob(
     const { getTableById } = await import('@/lib/table/service')
     const { getRowById } = await import('@/lib/table/rows/service')
     const { pickNextEligibleGroupForRow } = await import('@/lib/table/workflow-columns')
+    const { readStampedCapabilitySubject } = await import('@/lib/table/rows/executions')
 
     let currentPayload = payload
     while (true) {
@@ -344,6 +345,13 @@ export async function executeWorkflowGroupCellJob(
         // Re-derive so a workflow group after an enrichment group doesn't keep a stale enrichmentId.
         enrichmentId: next.enrichmentId,
         executionId: generateId(),
+        /**
+         * The marker was stamped by whichever dispatch requested THIS cell,
+         * which is not necessarily the one that queued this carrier. Its gate
+         * belongs to the person who asked for it, so take the subject off the
+         * stamp rather than carrying our own into someone else's request.
+         */
+        capabilityGovernedUserId: await readStampedCapabilitySubject(rowId, next.id),
       }
     }
   } finally {
@@ -362,12 +370,14 @@ export async function runRowCascadeLoop(
   const { getTableById } = await import('@/lib/table/service')
   const { getRowById } = await import('@/lib/table/rows/service')
   const { pickNextEligibleGroupForRow } = await import('@/lib/table/workflow-columns')
+  const { readStampedCapabilitySubject } = await import('@/lib/table/rows/executions')
 
   let currentGroupId = payload.groupId
   let currentWorkflowId = payload.workflowId
   // Fresh executionId per iteration: SQL guard rejects writes whose id ≠
   // row.executions[gid].executionId, so we need a new claim per group.
   let currentExecutionId = payload.executionId
+  let currentCapabilityGovernedUserId = payload.capabilityGovernedUserId ?? null
 
   while (true) {
     if (signal?.aborted) {
@@ -377,6 +387,7 @@ export async function runRowCascadeLoop(
           groupId: currentGroupId,
           workflowId: currentWorkflowId,
           executionId: currentExecutionId,
+          capabilityGovernedUserId: currentCapabilityGovernedUserId,
         },
         signal
       )
@@ -400,6 +411,7 @@ export async function runRowCascadeLoop(
         groupId: currentGroupId,
         workflowId: currentWorkflowId,
         executionId: currentExecutionId,
+        capabilityGovernedUserId: currentCapabilityGovernedUserId,
       },
       signal,
       freshTable,
@@ -416,6 +428,17 @@ export async function runRowCascadeLoop(
     if (!freshRow) break
     const next = pickNextEligibleGroupForRow(freshTable, freshRow, currentGroupId)
     if (!next) break
+    const nextExec = freshRow.executions?.[next.id]
+    /**
+     * A dep-fill cascade stays under the subject that started it. A group
+     * carrying an unclaimed pre-stamp is a different thing — an explicit
+     * request from another dispatch that this cascade is draining — so it runs
+     * under the subject stamped with that request.
+     */
+    currentCapabilityGovernedUserId =
+      nextExec?.status === 'pending' && nextExec.executionId == null
+        ? await readStampedCapabilitySubject(rowId, next.id)
+        : currentCapabilityGovernedUserId
     currentGroupId = next.id
     currentWorkflowId = next.workflowId
     currentExecutionId = generateId()

apps/sim/lib/table/dispatcher.ts

@@ -815,6 +815,15 @@ async function stampQueuedForBatch(
             jobId: null,
             workflowId: runOpts.workflowId,
             error: null,
+            /**
+             * The marker outlives this dispatch's own worker: a cell task that
+             * finds the row's cascade lock held bails, and whoever owns the lock
+             * drains this marker instead. Persisting the subject is what makes
+             * that drain run under the person who requested THIS cell rather
+             * than under the owner's — a different dispatch, and often an
+             * actorless auto-fire with no gate at all.
+             */
+            capabilityGovernedUserId: runOpts.capabilityGovernedUserId ?? null,
           },
         }
       )

apps/sim/lib/table/prestamp-governed-subject.test.ts

@@ -0,0 +1,85 @@
+/**
+ * @vitest-environment node
+ */
+import { dbChainMockFns, resetDbChainMock } from '@sim/testing'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const mocks = vi.hoisted(() => ({
+  getTableById: vi.fn(),
+  writeWorkflowGroupState: vi.fn(),
+  batchEnqueueAndWait: vi.fn(),
+}))
+
+vi.mock('@/lib/table/events', () => ({ appendTableEvent: vi.fn() }))
+vi.mock('@/lib/table/service', () => ({ getTableById: mocks.getTableById }))
+vi.mock('@/lib/table/cell-write', () => ({
+  writeWorkflowGroupState: mocks.writeWorkflowGroupState,
+}))
+vi.mock('@/lib/billing/core/billing-attribution', () => ({
+  assertBillingAttributionSnapshot: (snapshot: unknown) => snapshot,
+  resolveBillingAttribution: async () => ({ actorUserId: 'billing-owner' }),
+  resolveSystemBillingAttribution: async () => ({ actorUserId: null }),
+}))
+vi.mock('@/lib/core/async-jobs/config', () => ({
+  getJobQueue: async () => ({ batchEnqueueAndWait: mocks.batchEnqueueAndWait }),
+}))
+
+import { dispatcherStep } from '@/lib/table/dispatcher'
+
+const GROUP = { id: 'group-1', workflowId: 'workflow-1', outputs: [] }
+
+const DISPATCH = {
+  id: 'tdsp_1',
+  tableId: 'table-1',
+  workspaceId: 'workspace-1',
+  requestId: 'req-1',
+  mode: 'incomplete',
+  scope: { groupIds: ['group-1'] },
+  status: 'dispatching',
+  cursor: -1,
+  limit: null,
+  processedCount: 0,
+  isManualRun: true,
+  triggeredByUserId: 'billing-owner',
+  capabilityGovernedUserId: 'requesting-member',
+  requestedAt: new Date('2026-08-21T15:00:00.000Z'),
+  completedAt: null,
+  cancelledAt: null,
+}
+
+describe('the dispatcher pre-stamp', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetDbChainMock()
+    mocks.getTableById.mockResolvedValue({
+      id: 'table-1',
+      workspaceId: 'workspace-1',
+      schema: { columns: [], workflowGroups: [GROUP] },
+    })
+    mocks.writeWorkflowGroupState.mockResolvedValue('wrote')
+    dbChainMockFns.limit
+      .mockResolvedValueOnce([DISPATCH])
+      .mockResolvedValueOnce([{ id: 'row-1', tableId: 'table-1', position: 0, data: {} }])
+      .mockResolvedValueOnce([DISPATCH])
+  })
+
+  /**
+   * The marker outlives its own worker: a cell task that finds the row's
+   * cascade lock held bails, and the lock owner drains the marker. Without the
+   * subject on the stamp, that drain runs the request under the owner's
+   * subject — a different dispatch, often an ungated auto-fire.
+   */
+  it('stamps the dispatch’s governed subject onto every cell it queues', async () => {
+    await dispatcherStep('tdsp_1')
+
+    expect(mocks.writeWorkflowGroupState).toHaveBeenCalledWith(
+      expect.anything(),
+      expect.objectContaining({
+        executionState: expect.objectContaining({
+          status: 'pending',
+          capabilityGovernedUserId: 'requesting-member',
+        }),
+      })
+    )
+  })
+})

apps/sim/lib/table/rows/executions.test.ts

@@ -33,6 +33,48 @@ describe('writeExecutionsPatch guards', () => {
     resetDbChainMock()
   })
 
+  /**
+   * The dispatcher's `pending` marker is drained by whichever worker owns the
+   * row's cascade lock, which may belong to another dispatch entirely. Storing
+   * the requesting subject with the marker is what lets that drain run under
+   * the person who asked rather than under the owner's own subject.
+   */
+  it('persists the pre-stamp’s governed subject on both the insert and the upsert', async () => {
+    await writeExecutionsPatch(
+      dbChainMock.db as unknown as Parameters<typeof writeExecutionsPatch>[0],
+      'table-1',
+      'row-1',
+      {
+        'group-1': {
+          ...EXECUTION_STATE,
+          status: 'pending',
+          executionId: null,
+          capabilityGovernedUserId: 'requesting-member',
+        },
+      }
+    )
+
+    const values = dbChainMockFns.values.mock.calls[0]?.[0] as Record<string, unknown>
+    expect(values.capabilityGovernedUserId).toBe('requesting-member')
+    const conflict = dbChainMockFns.onConflictDoUpdate.mock.calls[0]?.[0] as {
+      set: Record<string, unknown>
+    }
+    expect(conflict.set.capabilityGovernedUserId).toBe('requesting-member')
+  })
+
+  /** A write that names no subject clears it — only an unclaimed marker is read. */
+  it('writes null for a state that carries no subject', async () => {
+    await writeExecutionsPatch(
+      dbChainMock.db as unknown as Parameters<typeof writeExecutionsPatch>[0],
+      'table-1',
+      'row-1',
+      { 'group-1': EXECUTION_STATE }
+    )
+
+    const values = dbChainMockFns.values.mock.calls[0]?.[0] as Record<string, unknown>
+    expect(values.capabilityGovernedUserId).toBeNull()
+  })
+
   it('rejects a worker write when the atomic stale-or-cancel predicate returns no row', async () => {
     dbChainMockFns.returning.mockResolvedValueOnce([])