fix(realtime): keep debounced subblock saves in order

Allow one flush per subblock while newer edits coalesce in a separate
pending batch. Start the next ready batch after the active save completes,
including failures, so a slow older lookup cannot overwrite a newer edit.

Cover delayed lookups, coalescing, failure recovery, confirmation order,
and independent subblock saves with regression tests.

Author: j15z

apps/realtime/src/handlers/subblocks.test.ts

@@ -63,6 +63,17 @@ function setup() {
 const value = [{ usageControlExpression: 'force' }]
 const update = { blockId: 'agent-1', subblockId: 'tools', value, operationId: 'op-1', timestamp: 1 }
 
+function holdNextWorkflowLookup() {
+  let finish: (error?: Error) => void = () => {}
+  const result = new Promise<Array<{ id: string }>>((resolve, reject) => {
+    finish = (error) => (error ? reject(error) : resolve([{ id: 'workflow-1' }]))
+  })
+  mockSelect.mockReturnValueOnce({
+    from: () => ({ where: () => ({ limit: () => result }) }),
+  })
+  return finish
+}
+
 describe('debounced variable permission writes', () => {
   beforeEach(() => {
     vi.useFakeTimers()
@@ -145,4 +156,94 @@ describe('debounced variable permission writes', () => {
       expect.objectContaining({ operationId: 'op-2' })
     )
   })
+
+  it('preserves save and confirmation order when the older workflow lookup stalls', async () => {
+    const { handlers, emit } = setup()
+    const finishLookup = holdNextWorkflowLookup()
+    const newerValue = [{ usageControlExpression: 'none' }]
+
+    await handlers['subblock-update'](update)
+    await vi.advanceTimersByTimeAsync(25)
+    await handlers['subblock-update']({ ...update, operationId: 'op-2', value: newerValue })
+    await vi.advanceTimersByTimeAsync(25)
+    const writesBeforeRelease = mockSet.mock.calls.length
+    finishLookup()
+    await vi.advanceTimersByTimeAsync(0)
+
+    expect(writesBeforeRelease).toBe(0)
+    expect(
+      mockSet.mock.calls
+        .filter(([fields]) => fields.subBlocks)
+        .map(([fields]) => fields.subBlocks.tools.value)
+    ).toEqual([value, newerValue])
+    expect(
+      emit.mock.calls
+        .filter(([event]) => event === 'operation-confirmed')
+        .map(([, payload]) => payload.operationId)
+    ).toEqual(['op-1', 'op-2'])
+  })
+
+  it.each([false, true])(
+    'coalesces waiting edits and continues after an older failure: %s',
+    async (failOlder) => {
+      const { handlers, emit } = setup()
+      const finishLookup = holdNextWorkflowLookup()
+      const newestValue = [{ usageControlExpression: 'auto' }]
+
+      await handlers['subblock-update'](update)
+      await vi.advanceTimersByTimeAsync(25)
+      await handlers['subblock-update']({
+        ...update,
+        operationId: 'op-2',
+        value: [{ usageControlExpression: 'none' }],
+      })
+      await vi.advanceTimersByTimeAsync(25)
+      await handlers['subblock-update']({ ...update, operationId: 'op-3', value: newestValue })
+      await vi.advanceTimersByTimeAsync(25)
+      finishLookup(failOlder ? new Error('connection reset') : undefined)
+      await vi.advanceTimersByTimeAsync(0)
+
+      expect(
+        mockSet.mock.calls
+          .filter(([fields]) => fields.subBlocks)
+          .map(([fields]) => fields.subBlocks.tools.value)
+      ).toEqual(failOlder ? [newestValue] : [value, newestValue])
+      expect(
+        emit.mock.calls
+          .filter(([event]) => event === 'operation-confirmed')
+          .map(([, payload]) => payload.operationId)
+      ).toEqual(failOlder ? ['op-2', 'op-3'] : ['op-1', 'op-2', 'op-3'])
+      if (failOlder) {
+        expect(emit).toHaveBeenCalledWith(
+          'operation-failed',
+          expect.objectContaining({ operationId: 'op-1', retryable: true })
+        )
+      }
+    }
+  )
+
+  it('allows a different subblock to save while one subblock is stalled', async () => {
+    const { handlers, emit } = setup()
+    const finishLookup = holdNextWorkflowLookup()
+    await handlers['subblock-update'](update)
+    await vi.advanceTimersByTimeAsync(25)
+    await handlers['subblock-update']({
+      ...update,
+      subblockId: 'systemPrompt',
+      operationId: 'op-other',
+      value: 'hello',
+    })
+    await vi.advanceTimersByTimeAsync(25)
+    const confirmedBeforeRelease = emit.mock.calls
+      .filter(([event]) => event === 'operation-confirmed')
+      .map(([, payload]) => payload.operationId)
+    finishLookup()
+    await vi.advanceTimersByTimeAsync(0)
+
+    expect(confirmedBeforeRelease).toEqual(['op-other'])
+    expect(emit).toHaveBeenCalledWith(
+      'operation-confirmed',
+      expect.objectContaining({ operationId: 'op-1' })
+    )
+  })
 })

apps/realtime/src/handlers/subblocks.ts

@@ -19,12 +19,14 @@ const DEBOUNCE_INTERVAL_MS = 25
 type PendingSubblock = {
   latest: { blockId: string; subblockId: string; value: any; timestamp: number }
   timeout: NodeJS.Timeout
+  ready: boolean
   // Map operationId -> socketId to emit confirmations/failures to correct clients
   opToSocket: Map<string, string>
 }
 
 // Keyed by `${workflowId}:${blockId}:${subblockId}`
 const pendingSubblockUpdates = new Map<string, PendingSubblock>()
+const flushingSubblockUpdates = new Set<string>()
 
 /**
  * Cleans up pending updates for a disconnected socket.
@@ -192,24 +194,26 @@ export function setupSubblocksHandlers(socket: AuthenticatedSocket, roomManager:
       if (existing) {
         clearTimeout(existing.timeout)
         existing.latest = { blockId, subblockId, value, timestamp }
+        existing.ready = false
         if (operationId) existing.opToSocket.set(operationId, socket.id)
         existing.timeout = setTimeout(async () => {
-          pendingSubblockUpdates.delete(debouncedKey)
-          await flushSubblockUpdate(workflowId, existing, roomManager)
+          existing.ready = true
+          await flushReadySubblockUpdates(workflowId, debouncedKey, roomManager)
         }, DEBOUNCE_INTERVAL_MS)
       } else {
         const opToSocket = new Map<string, string>()
         if (operationId) opToSocket.set(operationId, socket.id)
         const timeout = setTimeout(async () => {
           const pending = pendingSubblockUpdates.get(debouncedKey)
           if (pending) {
-            pendingSubblockUpdates.delete(debouncedKey)
-            await flushSubblockUpdate(workflowId, pending, roomManager)
+            pending.ready = true
+            await flushReadySubblockUpdates(workflowId, debouncedKey, roomManager)
           }
         }, DEBOUNCE_INTERVAL_MS)
         pendingSubblockUpdates.set(debouncedKey, {
           latest: { blockId, subblockId, value, timestamp },
           timeout,
+          ready: false,
           opToSocket,
         })
       }
@@ -236,6 +240,26 @@ export function setupSubblocksHandlers(socket: AuthenticatedSocket, roomManager:
   })
 }
 
+/** Keep one save in progress per subblock while newer edits coalesce in a separate batch. */
+async function flushReadySubblockUpdates(
+  workflowId: string,
+  debouncedKey: string,
+  roomManager: IRoomManager
+) {
+  if (flushingSubblockUpdates.has(debouncedKey)) return
+  flushingSubblockUpdates.add(debouncedKey)
+  try {
+    let pending = pendingSubblockUpdates.get(debouncedKey)
+    while (pending?.ready) {
+      pendingSubblockUpdates.delete(debouncedKey)
+      await flushSubblockUpdate(workflowId, pending, roomManager)
+      pending = pendingSubblockUpdates.get(debouncedKey)
+    }
+  } finally {
+    flushingSubblockUpdates.delete(debouncedKey)
+  }
+}
+
 async function flushSubblockUpdate(
   workflowId: string,
   pending: PendingSubblock,