fix(tables): server-authoritative run badge, tail SSE from latest, harden Stop-all

apps/sim/app/api/table/[tableId]/dispatches/route.test.ts

@@ -0,0 +1,129 @@
+/**
+ * @vitest-environment node
+ */
+import { hybridAuthMockFns } from '@sim/testing'
+import { NextRequest } from 'next/server'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+import type { TableDefinition } from '@/lib/table'
+
+const { mockCheckAccess, mockListActiveDispatches, mockCountRunningCells } = vi.hoisted(() => ({
+  mockCheckAccess: vi.fn(),
+  mockListActiveDispatches: vi.fn(),
+  mockCountRunningCells: vi.fn(),
+}))
+
+vi.mock('@/lib/table/dispatcher', () => ({
+  listActiveDispatches: mockListActiveDispatches,
+  countRunningCells: mockCountRunningCells,
+}))
+vi.mock('@/app/api/table/utils', async () => {
+  const { NextResponse } = await import('next/server')
+  return {
+    checkAccess: mockCheckAccess,
+    accessError: (result: { status: number }) =>
+      NextResponse.json({ error: 'denied' }, { status: result.status }),
+  }
+})
+
+import { GET } from '@/app/api/table/[tableId]/dispatches/route'
+
+function buildTable(overrides: Partial<TableDefinition> = {}): TableDefinition {
+  return {
+    id: 'tbl_1',
+    name: 'People',
+    description: null,
+    schema: { columns: [] },
+    metadata: null,
+    rowCount: 0,
+    maxRows: 1_000_000,
+    workspaceId: 'workspace-1',
+    createdBy: 'user-1',
+    archivedAt: null,
+    createdAt: new Date(),
+    updatedAt: new Date(),
+    ...overrides,
+  }
+}
+
+function makeRequest(tableId = 'tbl_1') {
+  const req = new NextRequest(`http://localhost:3000/api/table/${tableId}/dispatches`)
+  return GET(req, { params: Promise.resolve({ tableId }) })
+}
+
+function buildDispatchRow(overrides: Record<string, unknown> = {}) {
+  return {
+    id: 'dispatch-1',
+    tableId: 'tbl_1',
+    workspaceId: 'workspace-1',
+    requestId: 'req-1',
+    mode: 'all',
+    scope: { groupIds: ['group-1'] },
+    status: 'dispatching',
+    cursor: 4,
+    limit: null,
+    isManualRun: true,
+    processedCount: 5,
+    ...overrides,
+  }
+}
+
+describe('GET /api/table/[tableId]/dispatches', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    hybridAuthMockFns.mockCheckSessionOrInternalAuth.mockResolvedValue({
+      success: true,
+      userId: 'user-1',
+      authType: 'session',
+    })
+    mockCheckAccess.mockResolvedValue({ ok: true, table: buildTable() })
+    mockListActiveDispatches.mockResolvedValue([])
+    mockCountRunningCells.mockResolvedValue({ byRowId: {}, hasRunning: false })
+  })
+
+  it('returns dispatches and the per-row running map, without a total field', async () => {
+    mockListActiveDispatches.mockResolvedValue([buildDispatchRow()])
+    mockCountRunningCells.mockResolvedValue({
+      byRowId: { 'row-1': 2, 'row-2': 1 },
+      hasRunning: true,
+    })
+
+    const response = await makeRequest()
+    const data = await response.json()
+
+    expect(response.status).toBe(200)
+    expect(data.data.dispatches).toEqual([
+      {
+        id: 'dispatch-1',
+        status: 'dispatching',
+        mode: 'all',
+        isManualRun: true,
+        cursor: 4,
+        scope: { groupIds: ['group-1'] },
+      },
+    ])
+    expect(data.data.runningByRowId).toEqual({ 'row-1': 2, 'row-2': 1 })
+    expect(data.data.hasRunning).toBe(true)
+    expect(data.data).not.toHaveProperty('runningCellCount')
+  })
+
+  it('includes unclaimed pre-stamps only while a dispatch is active', async () => {
+    mockListActiveDispatches.mockResolvedValue([buildDispatchRow()])
+    await makeRequest()
+    expect(mockCountRunningCells).toHaveBeenCalledWith('tbl_1', {
+      includeUnclaimedPreStamps: true,
+    })
+
+    mockListActiveDispatches.mockResolvedValue([])
+    await makeRequest()
+    expect(mockCountRunningCells).toHaveBeenLastCalledWith('tbl_1', {
+      includeUnclaimedPreStamps: false,
+    })
+  })
+
+  it('returns 401 when unauthenticated', async () => {
+    hybridAuthMockFns.mockCheckSessionOrInternalAuth.mockResolvedValue({ success: false })
+    const response = await makeRequest()
+    expect(response.status).toBe(401)
+    expect(mockListActiveDispatches).not.toHaveBeenCalled()
+  })
+})

apps/sim/app/api/table/[tableId]/dispatches/route.ts

@@ -5,7 +5,7 @@ import { parseRequest } from '@/lib/api/server'
 import { checkSessionOrInternalAuth } from '@/lib/auth/hybrid'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
-import { countActiveRunCells, listActiveDispatches } from '@/lib/table/dispatcher'
+import { countRunningCells, listActiveDispatches } from '@/lib/table/dispatcher'
 import { accessError, checkAccess } from '@/app/api/table/utils'
 
 const logger = createLogger('TableDispatchesAPI')
@@ -38,7 +38,12 @@ export const GET = withRouteHandler(async (request: NextRequest, { params }: Rou
     if (!result.ok) return accessError(result, requestId, tableId)
 
     const rows = await listActiveDispatches(tableId)
-    const running = await countActiveRunCells(tableId, rows)
+    // Unclaimed `pending` pre-stamps are real queued work while a dispatch is
+    // active; with none active they're abandoned orphans that would pin the
+    // "X running" badge above zero forever.
+    const { byRowId: runningByRowId, hasRunning } = await countRunningCells(tableId, {
+      includeUnclaimedPreStamps: rows.length > 0,
+    })
     const dispatches: ActiveDispatch[] = rows.map((r) => ({
       id: r.id,
       status: r.status as 'pending' | 'dispatching',
@@ -53,8 +58,8 @@ export const GET = withRouteHandler(async (request: NextRequest, { params }: Rou
       success: true,
       data: {
         dispatches,
-        runningCellCount: running.total,
-        runningByRowId: running.byRowId,
+        runningByRowId,
+        hasRunning,
       },
     })
   } catch (error) {

apps/sim/app/api/table/[tableId]/events/stream/route.ts

@@ -8,7 +8,11 @@ import { checkSessionOrInternalAuth } from '@/lib/auth/hybrid'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { SSE_HEADERS } from '@/lib/core/utils/sse'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
-import { readTableEventsSince, type TableEventEntry } from '@/lib/table/events'
+import {
+  getLatestTableEventId,
+  readTableEventsSince,
+  type TableEventEntry,
+} from '@/lib/table/events'
 import { accessError, checkAccess } from '@/app/api/table/utils'
 
 const logger = createLogger('TableEventStreamAPI')
@@ -26,10 +30,12 @@ interface RouteContext {
 
 /** GET /api/table/[tableId]/events/stream?from=<lastEventId>
  *
- *  SSE stream of cell-state transitions. Replay-on-reconnect via `from`.
+ *  SSE stream of cell-state transitions. Replay-on-reconnect via `from`;
+ *  absent `from` tails from the latest event id (fresh mount — the client has
+ *  just fetched current state, so replaying history would rewind it).
  *  Pruning (buffer cap exceeded or TTL expired) sends a `pruned` event and
  *  closes; the client responds with a full row-query refetch and reconnects
- *  from the new earliest. */
+ *  tailing from latest. */
 export const GET = withRouteHandler(async (req: NextRequest, context: RouteContext) => {
   const requestId = generateRequestId()
   const parsed = await parseRequest(tableEventStreamContract, req, context)
@@ -52,7 +58,7 @@ export const GET = withRouteHandler(async (req: NextRequest, context: RouteConte
 
   const stream = new ReadableStream<Uint8Array>({
     async start(controller) {
-      let lastEventId = fromEventId
+      let lastEventId = fromEventId ?? 0
       const deadline = Date.now() + MAX_STREAM_DURATION_MS
       let nextHeartbeatAt = Date.now() + HEARTBEAT_INTERVAL_MS
 
@@ -92,6 +98,12 @@ export const GET = withRouteHandler(async (req: NextRequest, context: RouteConte
       }
 
       try {
+        // No replay cursor → tail from the latest event id. Resolved inside
+        // the try so a Redis failure errors the stream (client reconnects
+        // with backoff) rather than silently replaying the whole buffer.
+        if (fromEventId === undefined) {
+          lastEventId = await getLatestTableEventId(tableId)
+        }
         // Initial replay from buffer.
         const initial = await readTableEventsSince(tableId, lastEventId)
         if (initial.status === 'pruned') {

apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/run-status-control/run-status-control.tsx

@@ -6,6 +6,10 @@ import { Loader, Square } from '@sim/emcn/icons'
 
 interface RunStatusControlProps {
   running: number
+  /** No cell has been claimed by a worker yet — everything counted is queued
+   *  or pending, so labeling it "running" would be dishonest. Renders
+   *  "Queueing" without a count instead. */
+  queueing: boolean
   onStopAll: () => void
   isStopping: boolean
 }
@@ -17,15 +21,22 @@ interface RunStatusControlProps {
  */
 export const RunStatusControl = memo(function RunStatusControl({
   running,
+  queueing,
   onStopAll,
   isStopping,
 }: RunStatusControlProps) {
   return (
     <div className='flex items-center gap-1.5'>
       <div className='flex items-center gap-1.5 px-1 text-[var(--text-tertiary)] text-caption'>
         <Loader animate className='size-[14px] shrink-0' />
-        <span className='tabular-nums'>{running}</span>
-        <span>running</span>
+        {queueing ? (
+          <span>Queueing</span>
+        ) : (
+          <>
+            <span className='tabular-nums'>{running}</span>
+            <span>running</span>
+          </>
+        )}
       </div>
       <Button
         variant='subtle'
@@ -34,7 +45,7 @@ export const RunStatusControl = memo(function RunStatusControl({
         disabled={isStopping}
       >
         <Square className='mr-1.5 size-[14px]' />
-        Stop all
+        {isStopping ? 'Stopping…' : 'Stop all'}
       </Button>
     </div>
   )