perf(db): optimize recurring query paths (#7014)

* perf(db): optimize recurring query paths * perf(logs): batch keyset export reads * fix(workspaces): type nullable member count targets * fix(db): harden query performance changes * fix(logs): guard export stream cancellation
2026-10-30 19:32:46 +08:00 · 2026-08-23 17:15:10 -07:00
parent 9cecf0b837
commit a20a5465ef
20 changed files with 21383 additions and 211 deletions
@@ -0,0 +1,239 @@
+/**
+ * @vitest-environment node
+ */
+import { workflowExecutionLogs } from '@sim/db/schema'
+import {
+  authMockFns,
+  createMockRequest,
+  dbChainMockFns,
+  queueTableRows,
+  resetDbChainMock,
+} from '@sim/testing'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const {
+  mockCheckWorkspaceAccess,
+  mockExpandFolderIdsWithDescendants,
+  mockMapWithConcurrency,
+  mockMaterializeExecutionDataForDisplay,
+} = vi.hoisted(() => ({
+  mockCheckWorkspaceAccess: vi.fn(),
+  mockExpandFolderIdsWithDescendants: vi.fn(),
+  mockMapWithConcurrency: vi.fn(),
+  mockMaterializeExecutionDataForDisplay: vi.fn(),
+}))
+
+vi.mock('@/lib/workspaces/permissions/utils', () => ({
+  checkWorkspaceAccess: mockCheckWorkspaceAccess,
+}))
+
+vi.mock('@/lib/logs/folder-expansion', () => ({
+  expandFolderIdsWithDescendants: mockExpandFolderIdsWithDescendants,
+}))
+
+vi.mock('@/lib/logs/execution/trace-store', () => ({
+  materializeExecutionDataForDisplay: mockMaterializeExecutionDataForDisplay,
+}))
+
+vi.mock('@/lib/core/utils/concurrency', () => ({
+  MATERIALIZE_CONCURRENCY: 20,
+  mapWithConcurrency: mockMapWithConcurrency,
+}))
+
+import { GET } from '@/app/api/logs/export/route'
+
+const mockGetSession = authMockFns.mockGetSession
+const STARTED_AT = new Date('2026-08-23T12:00:00.000Z')
+
+function makeRequest() {
+  return createMockRequest(
+    'GET',
+    undefined,
+    {},
+    'http://localhost:3000/api/logs/export?workspaceId=workspace-1'
+  )
+}
+
+function logRow(index: number, overrides: Record<string, unknown> = {}) {
+  const startedAt = new Date(STARTED_AT.getTime() - index * 1000)
+  return {
+    id: `log-${index.toString().padStart(4, '0')}`,
+    workflowId: 'workflow-1',
+    executionId: `execution-${index}`,
+    level: 'info',
+    trigger: 'manual',
+    startedAt,
+    startedAtCursor: startedAt.toISOString(),
+    endedAt: new Date(STARTED_AT.getTime() - index * 1000 + 500),
+    totalDurationMs: 500,
+    costTotal: '0.01',
+    executionData: { message: `message-${index}` },
+    workflowName: 'Workflow',
+    ...overrides,
+  }
+}
+
+function flattenConditions(condition: unknown): Array<Record<string, unknown>> {
+  if (!condition || typeof condition !== 'object') return []
+  const node = condition as Record<string, unknown>
+  if (Array.isArray(node.conditions)) {
+    return node.conditions.flatMap(flattenConditions)
+  }
+  return [node]
+}
+
+describe('GET /api/logs/export', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetDbChainMock()
+    mockGetSession.mockResolvedValue({ user: { id: 'user-1' } })
+    mockCheckWorkspaceAccess.mockResolvedValue({ hasAccess: true })
+    mockExpandFolderIdsWithDescendants.mockImplementation(
+      async (_workspaceId: string, folderIds: string | undefined) => folderIds
+    )
+    mockMaterializeExecutionDataForDisplay.mockImplementation(
+      async (executionData: Record<string, unknown> | null | undefined) => executionData ?? {}
+    )
+    mockMapWithConcurrency.mockImplementation(
+      async (
+        items: unknown[],
+        _limit: number,
+        mapper: (item: unknown, index: number) => Promise<unknown>
+      ) => Promise.all(items.map(mapper))
+    )
+  })
+
+  it('rejects unauthenticated exports before checking workspace access', async () => {
+    mockGetSession.mockResolvedValueOnce(null)
+
+    const response = await GET(makeRequest())
+
+    expect(response.status).toBe(401)
+    expect(mockCheckWorkspaceAccess).not.toHaveBeenCalled()
+    expect(dbChainMockFns.where).not.toHaveBeenCalled()
+    expect(mockMaterializeExecutionDataForDisplay).not.toHaveBeenCalled()
+  })
+
+  it('returns only the CSV header when workspace access is denied', async () => {
+    mockCheckWorkspaceAccess.mockResolvedValueOnce({ hasAccess: false })
+
+    const response = await GET(makeRequest())
+
+    expect(response.status).toBe(200)
+    expect(await response.text()).toBe(
+      'startedAt,level,workflow,trigger,durationMs,costTotal,workflowId,executionId,message,traceSpans\n'
+    )
+    expect(dbChainMockFns.where).not.toHaveBeenCalled()
+    expect(mockMaterializeExecutionDataForDisplay).not.toHaveBeenCalled()
+  })
+
+  it('materializes bounded chunks while preserving CSV row order', async () => {
+    queueTableRows(
+      workflowExecutionLogs,
+      Array.from({ length: 45 }, (_, index) => logRow(index))
+    )
+
+    const response = await GET(makeRequest())
+    const lines = (await response.text()).trimEnd().split('\n')
+
+    expect(response.status).toBe(200)
+    expect(mockMapWithConcurrency.mock.calls.map(([items]) => items.length)).toEqual([20, 20, 5])
+    expect(lines).toHaveLength(46)
+    expect(lines[1]).toContain('execution-0')
+    expect(lines.at(-1)).toContain('execution-44')
+  })
+
+  it('resumes full pages by startedAt and id without using OFFSET', async () => {
+    const firstPage = Array.from({ length: 100 }, (_, index) => logRow(index))
+    firstPage[99] = logRow(99, { startedAtCursor: '2026-08-23 11:58:21.000123' })
+    const last = firstPage.at(-1)!
+    const secondPage = [
+      logRow(100, {
+        id: 'log-0000-second',
+        startedAt: last.startedAt,
+        startedAtCursor: '2026-08-23 11:58:21.000122',
+      }),
+    ]
+    queueTableRows(workflowExecutionLogs, firstPage)
+    queueTableRows(workflowExecutionLogs, secondPage)
+
+    const response = await GET(makeRequest())
+    const lines = (await response.text()).trimEnd().split('\n')
+
+    expect(lines).toHaveLength(102)
+    expect(dbChainMockFns.offset).not.toHaveBeenCalled()
+    expect(dbChainMockFns.where).toHaveBeenCalledTimes(2)
+    expect(dbChainMockFns.orderBy).toHaveBeenNthCalledWith(
+      1,
+      expect.objectContaining({
+        type: 'desc',
+        column: workflowExecutionLogs.startedAt,
+      }),
+      expect.objectContaining({
+        type: 'desc',
+        column: workflowExecutionLogs.id,
+      })
+    )
+
+    const cursorConditions = flattenConditions(dbChainMockFns.where.mock.calls[1][0])
+    const timestampConditions = cursorConditions.filter(
+      (condition) => condition.left === workflowExecutionLogs.startedAt
+    )
+    expect(timestampConditions.map((condition) => condition.type)).toEqual(['lt', 'eq'])
+    for (const condition of timestampConditions) {
+      expect(condition.right).not.toBeInstanceOf(Date)
+      expect(condition.right).toEqual(
+        expect.objectContaining({ values: expect.arrayContaining([last.startedAtCursor]) })
+      )
+    }
+    expect(cursorConditions).toContainEqual(
+      expect.objectContaining({
+        type: 'lt',
+        left: workflowExecutionLogs.id,
+        right: last.id,
+      })
+    )
+  })
+
+  it('does not load the next database page until the current row is consumed', async () => {
+    queueTableRows(
+      workflowExecutionLogs,
+      Array.from({ length: 100 }, (_, index) => logRow(index))
+    )
+    queueTableRows(workflowExecutionLogs, [logRow(1)])
+
+    const response = await GET(makeRequest())
+    const reader = response.body!.getReader()
+
+    await reader.read()
+    expect(dbChainMockFns.where).not.toHaveBeenCalled()
+
+    await reader.read()
+    expect(dbChainMockFns.where).toHaveBeenCalledTimes(1)
+
+    await reader.cancel()
+    expect(dbChainMockFns.where).toHaveBeenCalledTimes(1)
+  })
+
+  it('stops a pending pull cleanly when the reader cancels', async () => {
+    queueTableRows(workflowExecutionLogs, [logRow(0)])
+    let resolveMaterialization: ((value: unknown[]) => void) | undefined
+    mockMapWithConcurrency.mockImplementationOnce(
+      () =>
+        new Promise((resolve) => {
+          resolveMaterialization = resolve
+        })
+    )
+
+    const response = await GET(makeRequest())
+    const reader = response.body!.getReader()
+    await reader.read()
+
+    const pendingRead = reader.read()
+    await vi.waitFor(() => expect(mockMapWithConcurrency).toHaveBeenCalledTimes(1))
+    const cancellation = reader.cancel()
+    resolveMaterialization?.([{ message: 'message-0' }])
+
+    await expect(Promise.all([pendingRead, cancellation])).resolves.toBeDefined()
+  })
+})
@@ -2,7 +2,7 @@ import { dbReplica } from '@sim/db'
 import { workflow, workflowExecutionLogs } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { getErrorMessage } from '@sim/utils/errors'
-import { and, desc, eq, sql } from 'drizzle-orm'
+import { and, desc, eq, lt, or, type SQL, sql } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import { getSession } from '@/lib/auth'
 import { MATERIALIZE_CONCURRENCY, mapWithConcurrency } from '@/lib/core/utils/concurrency'
@@ -14,9 +14,25 @@ import { expandFolderIdsWithDescendants } from '@/lib/logs/folder-expansion'
 import { checkWorkspaceAccess } from '@/lib/workspaces/permissions/utils'

 const logger = createLogger('LogsExportAPI')
+const LOG_EXPORT_PAGE_SIZE = 100

 export const revalidate = 0

+interface LogExportRow {
+  id: string
+  workflowId: string | null
+  executionId: string
+  level: string
+  trigger: string
+  startedAt: Date
+  startedAtCursor: string
+  endedAt: Date | null
+  totalDurationMs: number | null
+  costTotal: string | null
+  executionData: unknown
+  workflowName: string
+}
+
 export const GET = withRouteHandler(async (request: NextRequest) => {
  try {
    const session = await getSession()
@@ -35,6 +51,7 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
      level: workflowExecutionLogs.level,
      trigger: workflowExecutionLogs.trigger,
      startedAt: workflowExecutionLogs.startedAt,
+      startedAtCursor: sql<string>`${workflowExecutionLogs.startedAt}::text`,
      endedAt: workflowExecutionLogs.endedAt,
      totalDurationMs: workflowExecutionLogs.totalDurationMs,
      costTotal: workflowExecutionLogs.costTotal,
@@ -78,97 +95,118 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
    }

    const encoder = new TextEncoder()
-    const stream = new ReadableStream<Uint8Array>({
-      start: async (controller) => {
-        controller.enqueue(encoder.encode(`${header}\n`))
-        const pageSize = 1000
-        let offset = 0
-        try {
-          while (true) {
-            const rows = await dbReplica
-              .select(selectColumns)
-              .from(workflowExecutionLogs)
-              .leftJoin(workflow, eq(workflowExecutionLogs.workflowId, workflow.id))
-              .where(conditions)
-              .orderBy(desc(workflowExecutionLogs.startedAt))
-              .limit(pageSize)
-              .offset(offset)
-
-            if (!rows.length) break
-
-            // Heavy execution data may live in object storage; materialize per
-            // row with bounded concurrency so a 1000-row page doesn't fan out
-            // into 1000 simultaneous reads.
-            const materialized = await mapWithConcurrency(
-              rows as any[],
-              MATERIALIZE_CONCURRENCY,
-              (r) =>
-                materializeExecutionDataForDisplay(
-                  r.executionData as Record<string, unknown> | null,
-                  {
-                    workspaceId: params.workspaceId,
-                    workflowId: r.workflowId,
-                    executionId: r.executionId,
-                    userId: session.user.id,
-                  }
-                )
+    const csvChunks = (async function* () {
+      yield encoder.encode(`${header}\n`)
+      const pageSize = LOG_EXPORT_PAGE_SIZE
+      let cursor: { startedAt: string; id: string } | null = null
+      while (true) {
+        const cursorCondition: SQL | undefined = cursor
+          ? or(
+              lt(workflowExecutionLogs.startedAt, sql`${cursor.startedAt}::timestamp`),
+              and(
+                eq(workflowExecutionLogs.startedAt, sql`${cursor.startedAt}::timestamp`),
+                lt(workflowExecutionLogs.id, cursor.id)
+              )
            )
+          : undefined
+        const rows: LogExportRow[] = await dbReplica
+          .select(selectColumns)
+          .from(workflowExecutionLogs)
+          .leftJoin(workflow, eq(workflowExecutionLogs.workflowId, workflow.id))
+          .where(and(conditions, cursorCondition))
+          .orderBy(desc(workflowExecutionLogs.startedAt), desc(workflowExecutionLogs.id))
+          .limit(pageSize)

-            for (let j = 0; j < rows.length; j++) {
-              const r = rows[j] as any
-              const ed = materialized[j] as Record<string, any>
-              // A single malformed/unserializable row must not abort the whole CSV
-              // stream — derive the message/trace columns defensively and fall back
-              // to empty on error so the row's metadata still exports.
-              let message = ''
-              let tracesJson = ''
-              try {
-                if (ed) {
-                  if (ed.finalOutput)
-                    message =
-                      typeof ed.finalOutput === 'string'
-                        ? ed.finalOutput
-                        : JSON.stringify(ed.finalOutput)
-                  if (ed.message) message = ed.message
-                  if (ed.traceSpans) tracesJson = JSON.stringify(ed.traceSpans)
-                }
-              } catch (rowError) {
-                logger.warn('Skipping unserializable execution data for export row', {
-                  executionId: r.executionId,
-                  error: getErrorMessage(rowError),
-                })
+        if (!rows.length) break
+
+        for (let chunkStart = 0; chunkStart < rows.length; chunkStart += MATERIALIZE_CONCURRENCY) {
+          const chunk = rows.slice(chunkStart, chunkStart + MATERIALIZE_CONCURRENCY)
+          const materialized = await mapWithConcurrency(chunk, MATERIALIZE_CONCURRENCY, (row) =>
+            materializeExecutionDataForDisplay(
+              row.executionData as Record<string, unknown> | null,
+              {
+                workspaceId: params.workspaceId,
+                workflowId: row.workflowId,
+                executionId: row.executionId,
+                userId: session.user.id,
              }
-              const line = toCsvRow([
-                formatCsvValue(r.startedAt?.toISOString?.() || r.startedAt),
-                formatCsvValue(r.level),
-                formatCsvValue(r.workflowName),
-                formatCsvValue(r.trigger),
-                formatCsvValue(r.totalDurationMs ?? ''),
-                formatCsvValue(r.costTotal ?? ''),
-                formatCsvValue(r.workflowId ?? ''),
-                formatCsvValue(r.executionId ?? ''),
-                formatCsvValue(message),
-                formatCsvValue(tracesJson),
-              ])
-              controller.enqueue(encoder.encode(`${line}\n`))
-            }
+            )
+          )

-            offset += pageSize
+          for (let index = 0; index < chunk.length; index++) {
+            const row = chunk[index]
+            const executionData = materialized[index]
+            let message: unknown = ''
+            let tracesJson = ''
+            try {
+              if (executionData.finalOutput) {
+                message =
+                  typeof executionData.finalOutput === 'string'
+                    ? executionData.finalOutput
+                    : (JSON.stringify(executionData.finalOutput) ?? '')
+              }
+              if (executionData.message) message = executionData.message
+              if (executionData.traceSpans) {
+                tracesJson = JSON.stringify(executionData.traceSpans) ?? ''
+              }
+            } catch (rowError) {
+              logger.warn('Skipping unserializable execution data for export row', {
+                executionId: row.executionId,
+                error: getErrorMessage(rowError),
+              })
+            }
+            const line = toCsvRow([
+              formatCsvValue(row.startedAt),
+              formatCsvValue(row.level),
+              formatCsvValue(row.workflowName),
+              formatCsvValue(row.trigger),
+              formatCsvValue(row.totalDurationMs ?? ''),
+              formatCsvValue(row.costTotal ?? ''),
+              formatCsvValue(row.workflowId ?? ''),
+              formatCsvValue(row.executionId),
+              formatCsvValue(message),
+              formatCsvValue(tracesJson),
+            ])
+            yield encoder.encode(`${line}\n`)
          }
-          controller.close()
-        } catch (e: any) {
-          logger.error('Export stream error', { error: e?.message })
-          try {
-            controller.error(e)
-          } catch {}
        }
+
+        const last = rows.at(-1)
+        if (!last || rows.length < pageSize) break
+        cursor = { startedAt: last.startedAtCursor, id: last.id }
+      }
+    })()
+
+    let cancelled = false
+    const stream = new ReadableStream<Uint8Array>(
+      {
+        pull: async (controller) => {
+          try {
+            const next = await csvChunks.next()
+            if (cancelled) return
+            if (next.done) {
+              controller.close()
+              return
+            }
+            controller.enqueue(next.value)
+          } catch (error) {
+            if (cancelled) return
+            logger.error('Export stream error', { error: getErrorMessage(error) })
+            controller.error(error)
+          }
+        },
+        cancel: async () => {
+          cancelled = true
+          await csvChunks.return(undefined)
+        },
      },
-    })
+      { highWaterMark: 0 }
+    )

    const ts = new Date().toISOString().replace(/[:.]/g, '-')
    const filename = `logs-${ts}.csv`

-    return new NextResponse(stream as any, {
+    return new NextResponse(stream, {
      status: 200,
      headers: {
        'Content-Type': 'text/csv; charset=utf-8',
@@ -176,8 +214,8 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
        'Cache-Control': 'no-cache',
      },
    })
-  } catch (error: any) {
-    logger.error('Export error', { error: error?.message })
+  } catch (error) {
+    logger.error('Export error', { error: getErrorMessage(error) })
    return NextResponse.json({ error: 'Internal server error' }, { status: 500 })
  }
 })
@@ -11,6 +11,8 @@ const {
  mockGetPooledOrgCurrentPeriodCost,
  mockGetUserUsageLimit,
  mockIsOrganizationBillingBlocked,
+  mockComputeBillingPeriodUsageWithDailyRefresh,
+  mockGetOrgMemberRefreshBounds,
 } = vi.hoisted(() => ({
  mockGetBillingPeriodUsageCost: vi.fn(),
  mockGetOrgMemberUsageForBillingPeriod: vi.fn(),
@@ -18,6 +20,8 @@ const {
  mockGetPooledOrgCurrentPeriodCost: vi.fn(),
  mockGetUserUsageLimit: vi.fn(),
  mockIsOrganizationBillingBlocked: vi.fn(),
+  mockComputeBillingPeriodUsageWithDailyRefresh: vi.fn(),
+  mockGetOrgMemberRefreshBounds: vi.fn(),
 }))

 vi.mock('@/lib/billing/organizations/member-limits', () => ({
@@ -40,6 +44,11 @@ vi.mock('@/lib/billing/core/usage-log', () => ({
  getBillingPeriodUsageCost: mockGetBillingPeriodUsageCost,
 }))

+vi.mock('@/lib/billing/credits/daily-refresh', () => ({
+  computeBillingPeriodUsageWithDailyRefresh: mockComputeBillingPeriodUsageWithDailyRefresh,
+  getOrgMemberRefreshBounds: mockGetOrgMemberRefreshBounds,
+}))
+
 import {
  checkBillingBlocked,
  checkBillingEntityBlocked,
@@ -60,6 +69,11 @@ describe('checkUsageStatus', () => {
    setEnvFlags({ isHosted: true, isBillingEnabled: true })
    mockGetUserUsageLimit.mockResolvedValue(500)
    mockGetBillingPeriodUsageCost.mockResolvedValue(125)
+    mockComputeBillingPeriodUsageWithDailyRefresh.mockResolvedValue({
+      ledgerUsage: 125,
+      refreshConsumed: 25,
+    })
+    mockGetOrgMemberRefreshBounds.mockResolvedValue({})
  })

  it('reads reporting-period organization usage without loading the member roster', async () => {
@@ -97,6 +111,173 @@ describe('checkUsageStatus', () => {
    )
    expect(mockGetPooledOrgCurrentPeriodCost).not.toHaveBeenCalled()
  })
+
+  it('reads paid personal ledger usage and refresh from one snapshot', async () => {
+    const periodStart = new Date('2026-06-01T00:00:00.000Z')
+    const periodEnd = new Date('2026-07-01T00:00:00.000Z')
+    const subscription = {
+      referenceId: 'user-1',
+      plan: 'pro',
+      status: 'active',
+      seats: 1,
+      periodStart,
+      periodEnd,
+    }
+    dbChainMockFns.limit.mockResolvedValueOnce([{ currentPeriodCost: '20' }])
+
+    await expect(checkUsageStatus('user-1', subscription)).resolves.toMatchObject({
+      currentUsage: 120,
+      scope: 'user',
+    })
+
+    expect(mockComputeBillingPeriodUsageWithDailyRefresh).toHaveBeenCalledWith({
+      billingEntity: { type: 'user', id: 'user-1' },
+      billingPeriod: { start: periodStart, end: periodEnd },
+      userIds: ['user-1'],
+      refreshPeriodStart: periodStart,
+      refreshPeriodEnd: periodEnd,
+      planDollars: 20,
+    })
+    expect(mockGetBillingPeriodUsageCost).not.toHaveBeenCalled()
+  })
+
+  it('preserves the paid daily-refresh clamp for negative effective usage', async () => {
+    const periodStart = new Date('2026-06-01T00:00:00.000Z')
+    const periodEnd = new Date('2026-07-01T00:00:00.000Z')
+    const subscription = {
+      referenceId: 'user-1',
+      plan: 'pro',
+      status: 'active',
+      seats: 1,
+      periodStart,
+      periodEnd,
+    }
+    dbChainMockFns.limit.mockResolvedValueOnce([{ currentPeriodCost: '0' }])
+    mockComputeBillingPeriodUsageWithDailyRefresh.mockResolvedValueOnce({
+      ledgerUsage: -1,
+      refreshConsumed: 1,
+    })
+
+    await expect(checkUsageStatus('user-1', subscription)).resolves.toMatchObject({
+      currentUsage: 0,
+      scope: 'user',
+    })
+  })
+
+  it('keeps unpaid personal usage on the ledger-only query', async () => {
+    const periodStart = new Date('2026-06-01T00:00:00.000Z')
+    const periodEnd = new Date('2026-07-01T00:00:00.000Z')
+    const subscription = {
+      referenceId: 'user-1',
+      plan: 'free',
+      status: 'active',
+      seats: 1,
+      periodStart,
+      periodEnd,
+    }
+    dbChainMockFns.limit.mockResolvedValueOnce([{ currentPeriodCost: '20' }])
+
+    await expect(checkUsageStatus('user-1', subscription)).resolves.toMatchObject({
+      currentUsage: 145,
+      scope: 'user',
+    })
+
+    expect(mockGetBillingPeriodUsageCost).toHaveBeenCalledWith(
+      { type: 'user', id: 'user-1' },
+      { start: periodStart, end: periodEnd }
+    )
+    expect(mockComputeBillingPeriodUsageWithDailyRefresh).not.toHaveBeenCalled()
+  })
+
+  it('preserves negative ledger-only personal usage', async () => {
+    const periodStart = new Date('2026-06-01T00:00:00.000Z')
+    const periodEnd = new Date('2026-07-01T00:00:00.000Z')
+    const subscription = {
+      referenceId: 'user-1',
+      plan: 'free',
+      status: 'active',
+      seats: 1,
+      periodStart,
+      periodEnd,
+    }
+    dbChainMockFns.limit.mockResolvedValueOnce([{ currentPeriodCost: '0' }])
+    mockGetBillingPeriodUsageCost.mockResolvedValueOnce(-1)
+
+    await expect(checkUsageStatus('user-1', subscription)).resolves.toMatchObject({
+      currentUsage: -1,
+      scope: 'user',
+    })
+
+    expect(mockComputeBillingPeriodUsageWithDailyRefresh).not.toHaveBeenCalled()
+  })
+
+  it('combines paid organization ledger usage with bounded member refresh', async () => {
+    const periodStart = new Date('2026-06-01T00:00:00.000Z')
+    const periodEnd = new Date('2026-07-01T00:00:00.000Z')
+    const userStart = new Date('2026-06-10T00:00:00.000Z')
+    const subscription = {
+      referenceId: 'org-1',
+      plan: 'team',
+      status: 'active',
+      seats: 2,
+      periodStart,
+      periodEnd,
+    }
+    mockGetPooledOrgCurrentPeriodCost.mockResolvedValue({
+      memberIds: ['user-1', 'user-2'],
+      currentPeriodCost: 20,
+    })
+    mockGetOrgMemberRefreshBounds.mockResolvedValue({ 'user-2': { userStart } })
+    mockComputeBillingPeriodUsageWithDailyRefresh.mockResolvedValue({
+      ledgerUsage: 100,
+      refreshConsumed: 10,
+    })
+
+    await expect(checkUsageStatus('user-1', subscription)).resolves.toMatchObject({
+      currentUsage: 110,
+      scope: 'organization',
+      organizationId: 'org-1',
+    })
+
+    expect(mockComputeBillingPeriodUsageWithDailyRefresh).toHaveBeenCalledWith({
+      billingEntity: { type: 'organization', id: 'org-1' },
+      billingPeriod: expect.objectContaining({
+        start: periodStart,
+        end: periodEnd,
+        source: 'stripe',
+      }),
+      userIds: ['user-1', 'user-2'],
+      refreshPeriodStart: periodStart,
+      refreshPeriodEnd: periodEnd,
+      planDollars: expect.any(Number),
+      seats: 2,
+      userBounds: { 'user-2': { userStart } },
+    })
+    expect(mockGetBillingPeriodUsageCost).not.toHaveBeenCalled()
+  })
+
+  it('returns ledger usage without refresh when an organization has no members', async () => {
+    const periodStart = new Date('2026-06-01T00:00:00.000Z')
+    const periodEnd = new Date('2026-07-01T00:00:00.000Z')
+    const subscription = {
+      referenceId: 'org-1',
+      plan: 'team',
+      status: 'active',
+      seats: 1,
+      periodStart,
+      periodEnd,
+    }
+    mockGetPooledOrgCurrentPeriodCost.mockResolvedValue({ memberIds: [], currentPeriodCost: 0 })
+
+    await expect(checkUsageStatus('user-1', subscription)).resolves.toMatchObject({
+      currentUsage: 125,
+      scope: 'organization',
+    })
+
+    expect(mockGetBillingPeriodUsageCost).toHaveBeenCalledTimes(1)
+    expect(mockComputeBillingPeriodUsageWithDailyRefresh).not.toHaveBeenCalled()
+    expect(mockGetOrgMemberRefreshBounds).not.toHaveBeenCalled()
+  })
 })

 describe('checkBillingBlocked', () => {
@@ -20,7 +20,7 @@ import {
 } from '@/lib/billing/core/usage-log'
 import { dollarsToCredits } from '@/lib/billing/credits/conversion'
 import {
-  computeDailyRefreshConsumed,
+  computeBillingPeriodUsageWithDailyRefresh,
  getOrgMemberRefreshBounds,
 } from '@/lib/billing/credits/daily-refresh'
 import {
@@ -64,19 +64,49 @@ async function computePooledOrgUsage(
      anchorDate: null,
      interval: null,
    }
-  const ledgerUsage = await getBillingPeriodUsageCost(
-    { type: 'organization', id: organizationId },
-    billingPeriod
-  )
-
  if (billingPeriod.source === 'reporting') {
+    const ledgerUsage = await getBillingPeriodUsageCost(
+      { type: 'organization', id: organizationId },
+      billingPeriod
+    )
    return ledgerUsage
  }

  const { memberIds, currentPeriodCost } = await getPooledOrgCurrentPeriodCost(organizationId)
-  if (memberIds.length === 0) return ledgerUsage
+  if (memberIds.length === 0) {
+    return getBillingPeriodUsageCost({ type: 'organization', id: organizationId }, billingPeriod)
+  }

-  return applyOrgRefresh(organizationId, sub, currentPeriodCost + ledgerUsage, memberIds)
+  if (!isPaid(sub.plan) || !sub.periodStart) {
+    const ledgerUsage = await getBillingPeriodUsageCost(
+      { type: 'organization', id: organizationId },
+      billingPeriod
+    )
+    return currentPeriodCost + ledgerUsage
+  }
+
+  const planDollars = getPlanTierDollars(sub.plan)
+  if (planDollars <= 0) {
+    const ledgerUsage = await getBillingPeriodUsageCost(
+      { type: 'organization', id: organizationId },
+      billingPeriod
+    )
+    return currentPeriodCost + ledgerUsage
+  }
+
+  const userBounds = await getOrgMemberRefreshBounds(organizationId, sub.periodStart)
+  const { ledgerUsage, refreshConsumed } = await computeBillingPeriodUsageWithDailyRefresh({
+    billingEntity: { type: 'organization', id: organizationId },
+    billingPeriod,
+    userIds: memberIds,
+    refreshPeriodStart: sub.periodStart,
+    refreshPeriodEnd: sub.periodEnd ?? null,
+    planDollars,
+    seats: sub.seats || 1,
+    userBounds: Object.keys(userBounds).length > 0 ? userBounds : undefined,
+  })
+
+  return Math.max(0, currentPeriodCost + ledgerUsage - refreshConsumed)
 }

 /**
@@ -150,21 +180,32 @@ export async function checkUsageStatus(
      (sub?.periodStart && sub.periodEnd
        ? { start: sub.periodStart, end: sub.periodEnd }
        : defaultBillingPeriod())
-    const ledgerUsage = await getBillingPeriodUsageCost({ type: 'user', id: userId }, billingPeriod)
-    let currentUsage = toNumber(toDecimal(statsRecords[0].currentPeriodCost)) + ledgerUsage
+    let ledgerUsage: number
+    let refreshConsumed = 0
+    let appliedDailyRefresh = false
    if (sub && isPaid(sub.plan) && sub.periodStart) {
      const planDollars = getPlanTierDollars(sub.plan)
      if (planDollars > 0) {
-        const refresh = await computeDailyRefreshConsumed({
-          userIds: [userId],
-          periodStart: sub.periodStart,
-          periodEnd: sub.periodEnd ?? null,
-          planDollars,
+        const usage = await computeBillingPeriodUsageWithDailyRefresh({
          billingEntity: { type: 'user', id: userId },
+          billingPeriod,
+          userIds: [userId],
+          refreshPeriodStart: sub.periodStart,
+          refreshPeriodEnd: sub.periodEnd ?? null,
+          planDollars,
        })
-        currentUsage = Math.max(0, currentUsage - refresh)
+        ledgerUsage = usage.ledgerUsage
+        refreshConsumed = usage.refreshConsumed
+        appliedDailyRefresh = true
+      } else {
+        ledgerUsage = await getBillingPeriodUsageCost({ type: 'user', id: userId }, billingPeriod)
      }
+    } else {
+      ledgerUsage = await getBillingPeriodUsageCost({ type: 'user', id: userId }, billingPeriod)
    }
+    const usageBeforeRefresh =
+      toNumber(toDecimal(statsRecords[0].currentPeriodCost)) + ledgerUsage - refreshConsumed
+    const currentUsage = appliedDailyRefresh ? Math.max(0, usageBeforeRefresh) : usageBeforeRefresh

    return buildUsageData({ currentUsage, limit, scope, organizationId })
  } catch (error) {
@@ -190,42 +231,6 @@ export async function checkUsageStatus(
  }
 }

-async function applyOrgRefresh(
-  organizationId: string,
-  sub: {
-    plan: string | null
-    seats: number | null
-    periodStart: Date | null
-    periodEnd: Date | null
-  },
-  currentUsage: number,
-  preloadedMemberIds?: string[]
-): Promise<number> {
-  if (!isPaid(sub.plan) || !sub.periodStart) {
-    return currentUsage
-  }
-
-  const memberIds =
-    preloadedMemberIds ?? (await getPooledOrgCurrentPeriodCost(organizationId)).memberIds
-  if (memberIds.length === 0) return currentUsage
-
-  const planDollars = getPlanTierDollars(sub.plan)
-  if (planDollars <= 0) return currentUsage
-
-  const userBounds = await getOrgMemberRefreshBounds(organizationId, sub.periodStart)
-  const refresh = await computeDailyRefreshConsumed({
-    userIds: memberIds,
-    periodStart: sub.periodStart,
-    periodEnd: sub.periodEnd ?? null,
-    planDollars,
-    seats: sub.seats || 1,
-    userBounds: Object.keys(userBounds).length > 0 ? userBounds : undefined,
-    billingEntity: { type: 'organization', id: organizationId },
-  })
-
-  return Math.max(0, currentUsage - refresh)
-}
-
 function buildUsageData(params: {
  currentUsage: number
  limit: number
@@ -1,7 +1,7 @@
 /**
 * @vitest-environment node
 */
-import { dbChainMockFns, drizzleOrmMock } from '@sim/testing'
+import { dbChainMockFns, drizzleOrmMock, schemaMock } from '@sim/testing'
 import { beforeEach, describe, expect, it, vi } from 'vitest'

 vi.mock('drizzle-orm', () => {
@@ -21,10 +21,98 @@ vi.mock('@/lib/billing/constants', () => ({
 }))

 import {
+  computeBillingPeriodUsageWithDailyRefresh,
  computeDailyRefreshConsumed,
  getDailyRefreshDollars,
 } from '@/lib/billing/credits/daily-refresh'

+describe('computeBillingPeriodUsageWithDailyRefresh', () => {
+  const periodStart = new Date('2026-03-01T00:00:00.000Z')
+  const periodEnd = new Date('2026-04-01T00:00:00.000Z')
+
+  beforeEach(() => {
+    vi.clearAllMocks()
+  })
+
+  it('keeps the exact ledger end bound while computing refresh from daily buckets', async () => {
+    dbChainMockFns.groupBy.mockResolvedValueOnce([
+      { ledgerTotal: '12.50', refreshDayTotal: '0.50' },
+      { ledgerTotal: '12.50', refreshDayTotal: '0.10' },
+    ])
+
+    await expect(
+      computeBillingPeriodUsageWithDailyRefresh({
+        billingEntity: { type: 'organization', id: 'org-1' },
+        billingPeriod: { start: periodStart, end: periodEnd },
+        userIds: ['user-1'],
+        refreshPeriodStart: periodStart,
+        refreshPeriodEnd: periodEnd,
+        planDollars: 25,
+      })
+    ).resolves.toEqual({ ledgerUsage: 12.5, refreshConsumed: 0.35 })
+
+    expect(drizzleOrmMock.eq).toHaveBeenCalledWith(
+      schemaMock.usageLog.billingPeriodStart,
+      periodStart
+    )
+    expect(drizzleOrmMock.eq).toHaveBeenCalledWith(schemaMock.usageLog.billingPeriodEnd, periodEnd)
+  })
+
+  it('uses the reporting time range for the ledger while retaining captured-period refresh', async () => {
+    const reportingStart = new Date('2026-01-01T00:00:00.000Z')
+    const reportingEnd = new Date('2027-01-01T00:00:00.000Z')
+    dbChainMockFns.groupBy.mockResolvedValueOnce([
+      { ledgerTotal: '20.00', refreshDayTotal: '0.20' },
+    ])
+
+    await computeBillingPeriodUsageWithDailyRefresh({
+      billingEntity: { type: 'user', id: 'user-1' },
+      billingPeriod: {
+        start: reportingStart,
+        end: reportingEnd,
+        source: 'reporting',
+      },
+      userIds: ['user-1'],
+      refreshPeriodStart: periodStart,
+      refreshPeriodEnd: periodEnd,
+      planDollars: 25,
+    })
+
+    expect(drizzleOrmMock.gte).toHaveBeenCalledWith(schemaMock.usageLog.createdAt, reportingStart)
+    expect(drizzleOrmMock.lt).toHaveBeenCalledWith(schemaMock.usageLog.createdAt, reportingEnd)
+    expect(drizzleOrmMock.eq).toHaveBeenCalledWith(
+      schemaMock.usageLog.billingPeriodStart,
+      periodStart
+    )
+    expect(drizzleOrmMock.eq).not.toHaveBeenCalledWith(
+      schemaMock.usageLog.billingPeriodEnd,
+      reportingEnd
+    )
+  })
+
+  it('preserves a bounded user refresh window without narrowing the ledger total', async () => {
+    const userStart = new Date('2026-03-10T00:00:00.000Z')
+    const userEnd = new Date('2026-03-20T00:00:00.000Z')
+    dbChainMockFns.groupBy.mockResolvedValueOnce([
+      { ledgerTotal: '30.00', refreshDayTotal: '0.25' },
+    ])
+
+    await computeBillingPeriodUsageWithDailyRefresh({
+      billingEntity: { type: 'organization', id: 'org-1' },
+      billingPeriod: { start: periodStart, end: periodEnd },
+      userIds: ['bounded-user'],
+      refreshPeriodStart: periodStart,
+      refreshPeriodEnd: periodEnd,
+      planDollars: 25,
+      userBounds: { 'bounded-user': { userStart, userEnd } },
+    })
+
+    expect(drizzleOrmMock.eq).toHaveBeenCalledWith(schemaMock.usageLog.userId, 'bounded-user')
+    expect(drizzleOrmMock.gte).toHaveBeenCalledWith(schemaMock.usageLog.createdAt, userStart)
+    expect(drizzleOrmMock.lt).toHaveBeenCalledWith(schemaMock.usageLog.createdAt, userEnd)
+  })
+})
+
 describe('computeDailyRefreshConsumed', () => {
  beforeEach(() => {
    vi.clearAllMocks()
@@ -16,6 +16,7 @@ import { member, usageLog, userStats } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { and, eq, gte, inArray, isNull, lt, lte, or, sql, sum } from 'drizzle-orm'
 import { DAILY_REFRESH_RATE } from '@/lib/billing/constants'
+import type { BillingEntity, UsageQueryPeriod } from '@/lib/billing/core/usage-log'
 import type { DbClient } from '@/lib/db/types'

 const logger = createLogger('DailyRefresh')
@@ -34,6 +35,143 @@ export interface PerUserBounds {
  userEnd?: Date | null
 }

+interface BillingPeriodUsageWithDailyRefreshParams {
+  billingEntity: BillingEntity
+  billingPeriod: UsageQueryPeriod
+  userIds: string[]
+  refreshPeriodStart: Date
+  refreshPeriodEnd?: Date | null
+  planDollars: number
+  seats?: number
+  userBounds?: Record<string, PerUserBounds>
+}
+
+/**
+ * Reads the exact ledger total and the daily-refresh buckets from one snapshot.
+ *
+ * The two aggregates intentionally keep different predicates. Ledger totals use
+ * both captured period bounds (or a reporting-time window), while refresh uses
+ * the captured period start, eligible users, and per-user time bounds.
+ */
+export async function computeBillingPeriodUsageWithDailyRefresh(
+  params: BillingPeriodUsageWithDailyRefreshParams,
+  executor: DbClient = db
+): Promise<{ ledgerUsage: number; refreshConsumed: number }> {
+  const {
+    billingEntity,
+    billingPeriod,
+    userIds,
+    refreshPeriodStart,
+    refreshPeriodEnd,
+    planDollars,
+    seats = 1,
+    userBounds,
+  } = params
+  const now = new Date()
+  const cap = refreshPeriodEnd && refreshPeriodEnd < now ? refreshPeriodEnd : now
+  const dailyRefreshDollars = planDollars * DAILY_REFRESH_RATE * seats
+  const eligibleUserIds = new Set(userIds)
+  const unboundedUsers = userBounds ? userIds.filter((id) => !(id in userBounds)) : userIds
+  const boundedClauses = userBounds
+    ? Object.entries(userBounds).flatMap(([userId, bounds]) => {
+        if (!eligibleUserIds.has(userId)) return []
+        const effectiveStart =
+          bounds.userStart && bounds.userStart > refreshPeriodStart
+            ? bounds.userStart
+            : refreshPeriodStart
+        const effectiveEnd = bounds.userEnd && bounds.userEnd < cap ? bounds.userEnd : cap
+        if (effectiveEnd <= effectiveStart) return []
+        return [
+          and(
+            eq(usageLog.userId, userId),
+            gte(usageLog.createdAt, effectiveStart),
+            lt(usageLog.createdAt, effectiveEnd)
+          ),
+        ]
+      })
+    : []
+  const refreshUserFilters =
+    cap > refreshPeriodStart
+      ? [
+          ...(unboundedUsers.length > 0
+            ? [
+                and(
+                  inArray(usageLog.userId, unboundedUsers),
+                  gte(usageLog.createdAt, refreshPeriodStart),
+                  lt(usageLog.createdAt, cap)
+                ),
+              ]
+            : []),
+          ...boundedClauses,
+        ]
+      : []
+  const refreshFilter =
+    refreshUserFilters.length > 0
+      ? and(
+          eq(usageLog.billingPeriodStart, refreshPeriodStart),
+          refreshUserFilters.length === 1 ? refreshUserFilters[0] : or(...refreshUserFilters)
+        )
+      : sql<boolean>`false`
+  const ledgerPeriodFilter =
+    billingPeriod.source === 'reporting'
+      ? and(gte(usageLog.createdAt, billingPeriod.start), lt(usageLog.createdAt, billingPeriod.end))
+      : and(
+          eq(usageLog.billingPeriodStart, billingPeriod.start),
+          eq(usageLog.billingPeriodEnd, billingPeriod.end)
+        )
+
+  const sameCapturedPeriodStart =
+    billingPeriod.source !== 'reporting' &&
+    billingPeriod.start.getTime() === refreshPeriodStart.getTime()
+  const reportingWindowContainsRefresh =
+    billingPeriod.source === 'reporting' &&
+    refreshPeriodStart >= billingPeriod.start &&
+    cap <= billingPeriod.end
+  const scanFilter =
+    refreshUserFilters.length === 0
+      ? ledgerPeriodFilter
+      : sameCapturedPeriodStart
+        ? eq(usageLog.billingPeriodStart, billingPeriod.start)
+        : reportingWindowContainsRefresh
+          ? ledgerPeriodFilter
+          : or(ledgerPeriodFilter, refreshFilter)
+
+  const rows = await executor
+    .select({
+      dayIndex:
+        sql<number>`FLOOR((EXTRACT(EPOCH FROM ${usageLog.createdAt}) - ${Math.floor(refreshPeriodStart.getTime() / 1000)}) / 86400)`.as(
+          'day_index'
+        ),
+      ledgerTotal:
+        sql<string>`SUM(SUM(${usageLog.cost}) FILTER (WHERE ${ledgerPeriodFilter})) OVER ()`.as(
+          'ledger_total'
+        ),
+      refreshDayTotal: sql<string>`SUM(${usageLog.cost}) FILTER (WHERE ${refreshFilter})`.as(
+        'refresh_day_total'
+      ),
+    })
+    .from(usageLog)
+    .where(
+      and(
+        eq(usageLog.billingEntityType, billingEntity.type),
+        eq(usageLog.billingEntityId, billingEntity.id),
+        scanFilter
+      )
+    )
+    .groupBy(sql`day_index`)
+
+  let refreshConsumed = 0
+  for (const row of rows) {
+    const dayUsage = Number.parseFloat(row.refreshDayTotal ?? '0')
+    refreshConsumed += Math.min(dayUsage, dailyRefreshDollars)
+  }
+
+  return {
+    ledgerUsage: Number.parseFloat(rows[0]?.ledgerTotal ?? '0'),
+    refreshConsumed,
+  }
+}
+
 /**
 * Compute the total daily refresh credits consumed in the current billing period
 * using a single aggregating SQL query grouped by day offset.
@@ -15,6 +15,11 @@ const { findWorkspaceFileRecord, listAllWorkspaceFilesExecute, readWorkspaceFile
    readWorkspaceFileContentExecute: vi.fn(),
  }))

+const { isCustomBlocksEligible, listCustomBlocksWithInputsForWorkspace } = vi.hoisted(() => ({
+  isCustomBlocksEligible: vi.fn().mockResolvedValue(false),
+  listCustomBlocksWithInputsForWorkspace: vi.fn(),
+}))
+
 vi.mock('@/lib/copilot/tools/server/files/doc-render', () => ({
  // `odt` exposes the defensive missing-task branch independently from the extension guard.
  isRenderableDocExt: (ext: string) => ['docx', 'odt', 'pdf', 'pptx'].includes(ext.toLowerCase()),
@@ -33,12 +38,58 @@ vi.mock('@/lib/workspace-files/application/read-workspace-file-content', () => (
  readWorkspaceFileContent: { execute: readWorkspaceFileContentExecute },
 }))

+vi.mock('@/lib/workflows/custom-blocks/operations', () => ({
+  isCustomBlocksEligible,
+  listCustomBlocksWithInputsForWorkspace,
+}))
+
 import { WorkspaceVFS } from '@/lib/copilot/vfs/workspace-vfs'
 import { PayloadSizeLimitError } from '@/lib/core/utils/stream-limits'

 const MAX_DOC_READ_INPUT_BYTES = 50 * 1024 * 1024
 const MAX_DOCUMENT_PREVIEW_CODE_BYTES = 1024 * 1024

+interface TestableWorkspaceVFS {
+  loadCustomBlocks(workspaceId: string): Promise<unknown[]>
+}
+
+function customBlockLoader(vfs: WorkspaceVFS): TestableWorkspaceVFS {
+  return vfs as unknown as TestableWorkspaceVFS
+}
+
+describe('WorkspaceVFS custom block loading', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+  })
+
+  it('shares a successful custom block load within one VFS instance', async () => {
+    const blocks = [{ id: 'custom-block-1' }]
+    listCustomBlocksWithInputsForWorkspace.mockResolvedValueOnce(blocks)
+    const loader = customBlockLoader(new WorkspaceVFS())
+
+    const [first, second] = await Promise.all([
+      loader.loadCustomBlocks('workspace-1'),
+      loader.loadCustomBlocks('workspace-1'),
+    ])
+
+    expect(first).toBe(blocks)
+    expect(second).toBe(blocks)
+    expect(listCustomBlocksWithInputsForWorkspace).toHaveBeenCalledTimes(1)
+  })
+
+  it('retries a custom block load after failure', async () => {
+    const blocks = [{ id: 'custom-block-1' }]
+    listCustomBlocksWithInputsForWorkspace
+      .mockRejectedValueOnce(new Error('temporary failure'))
+      .mockResolvedValueOnce(blocks)
+    const loader = customBlockLoader(new WorkspaceVFS())
+
+    await expect(loader.loadCustomBlocks('workspace-1')).rejects.toThrow('temporary failure')
+    await expect(loader.loadCustomBlocks('workspace-1')).resolves.toBe(blocks)
+    expect(listCustomBlocksWithInputsForWorkspace).toHaveBeenCalledTimes(2)
+  })
+})
+
 function arrangeRenderRead({
  name = 'brief.pdf',
  size = 8,
@@ -163,7 +163,10 @@ import type {
  WorkspaceFileSecretProvenanceIdentity,
 } from '@/lib/uploads/contexts/workspace/workspace-file-secret-provenance'
 import { isImageFileType, resolveEffectiveMimeType } from '@/lib/uploads/utils/file-utils'
-import { listCustomBlocksWithInputsForWorkspace } from '@/lib/workflows/custom-blocks/operations'
+import {
+  type CustomBlockWithInputs,
+  listCustomBlocksWithInputsForWorkspace,
+} from '@/lib/workflows/custom-blocks/operations'
 import { getCustomToolById } from '@/lib/workflows/custom-tools/operations'
 import { checkNeedsRedeployment } from '@/lib/workflows/deployment-status'
 import {
@@ -669,6 +672,7 @@ export class WorkspaceVFS {
    Promise<Awaited<ReturnType<typeof loadWorkflowFromNormalizedTables>>>
  >()
  private deploymentCache = new Map<string, Promise<DeploymentData | null>>()
+  private customBlocksPromise: Promise<CustomBlockWithInputs[]> | undefined
  private _workspaceId = ''
  /**
   * Types of the org's CURRENT custom blocks (enabled + disabled — a disabled block
@@ -845,6 +849,7 @@ export class WorkspaceVFS {
    this.lazy = new Map()
    this.normalizedCache = new Map()
    this.deploymentCache = new Map()
+    this.customBlocksPromise = undefined
    this._customBlockTypes = null
    this._workspaceId = workspaceId

@@ -2394,7 +2399,7 @@ export class WorkspaceVFS {
    workspaceId: string
  ): Promise<NonNullable<WorkspaceMdData['customBlocks']>> {
    try {
-      const blocks = await listCustomBlocksWithInputsForWorkspace(workspaceId)
+      const blocks = await this.loadCustomBlocks(workspaceId)
      // Every current definition (incl. disabled) — the authoritative set used to
      // drop deleted-definition instances from workflow state (see loadNormalized).
      this._customBlockTypes = new Set(blocks.map((cb) => cb.type))
@@ -2431,6 +2436,18 @@ export class WorkspaceVFS {
    }
  }

+  /** Load the org's custom blocks once per VFS materialization. Failed loads remain retryable. */
+  private async loadCustomBlocks(workspaceId: string): Promise<CustomBlockWithInputs[]> {
+    const request = this.customBlocksPromise ?? listCustomBlocksWithInputsForWorkspace(workspaceId)
+    this.customBlocksPromise = request
+    try {
+      return await request
+    } catch (error) {
+      if (this.customBlocksPromise === request) this.customBlocksPromise = undefined
+      throw error
+    }
+  }
+
  /**
   * Materialize `account/` — the acting user's vantage: this workspace and
   * their role in it, the workspaces they can reach, who else is here, and
@@ -2583,7 +2600,7 @@ export class WorkspaceVFS {
      // names-only index need it, and each block's detail path must exist in
      // the key view for glob to list. Only the deployed graph stays lazy —
      // it is the expensive part and most turns never read it.
-      const orgBlocks = await listCustomBlocksWithInputsForWorkspace(workspaceId).catch((err) => {
+      const orgBlocks = await this.loadCustomBlocks(workspaceId).catch((err) => {
        logger.warn('Failed to list org custom blocks', {
          workspaceId,
          error: toError(err).message,
@@ -192,7 +192,7 @@ async function readKnowledgeBaseRows(
      deletedAt: knowledgeBase.deletedAt,
      workspaceId: knowledgeBase.workspaceId,
      folderId: knowledgeBase.folderId,
-      docCount: count(document.id),
+      docCount: count(document.knowledgeBaseId),
    })
    .from(knowledgeBase)
    .leftJoin(
@@ -882,7 +882,7 @@ export async function updateKnowledgeBase(
      deletedAt: knowledgeBase.deletedAt,
      workspaceId: knowledgeBase.workspaceId,
      folderId: knowledgeBase.folderId,
-      docCount: count(document.id),
+      docCount: count(document.knowledgeBaseId),
    })
    .from(knowledgeBase)
    .leftJoin(
@@ -941,7 +941,7 @@ export async function getKnowledgeBaseById(
      deletedAt: knowledgeBase.deletedAt,
      workspaceId: knowledgeBase.workspaceId,
      folderId: knowledgeBase.folderId,
-      docCount: count(document.id),
+      docCount: count(document.knowledgeBaseId),
    })
    .from(knowledgeBase)
    .leftJoin(
@@ -1,7 +1,13 @@
 /**
 * @vitest-environment node
 */
-import { describe, expect, it, vi } from 'vitest'
+import { queueTableRows, resetDbChainMock, schemaMock } from '@sim/testing'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { extractInputFieldsFromBlocks, loadDeployedWorkflowState } = vi.hoisted(() => ({
+  extractInputFieldsFromBlocks: vi.fn(),
+  loadDeployedWorkflowState: vi.fn(),
+}))

 vi.mock('@/lib/billing/core/subscription', () => ({
  isOrganizationOnEnterprisePlan: vi.fn(),
@@ -12,11 +18,11 @@ vi.mock('@/lib/core/config/feature-flags', () => ({
 }))

 vi.mock('@/lib/workflows/input-format', () => ({
-  extractInputFieldsFromBlocks: vi.fn(),
+  extractInputFieldsFromBlocks,
 }))

 vi.mock('@/lib/workflows/persistence/utils', () => ({
-  loadDeployedWorkflowState: vi.fn(),
+  loadDeployedWorkflowState,
 }))

 vi.mock('@/lib/workspaces/permissions/utils', () => ({
@@ -25,6 +31,7 @@ vi.mock('@/lib/workspaces/permissions/utils', () => ({

 import {
  CustomBlockValidationError,
+  listCustomBlocksWithInputs,
  publishCustomBlock,
  updateCustomBlock,
 } from '@/lib/workflows/custom-blocks/operations'
@@ -38,6 +45,83 @@ const publishParams = {
  description: '',
 }

+beforeEach(() => {
+  vi.clearAllMocks()
+  resetDbChainMock()
+})
+
+describe('custom block input hydration', () => {
+  it('passes the joined source workspace to deployed-state loading', async () => {
+    const block = {
+      id: 'custom-block-1',
+      organizationId: 'org-1',
+      workflowId: 'workflow-1',
+      type: 'custom_block_enrich',
+      name: 'Enrich Lead',
+      description: '',
+      iconUrl: null,
+      enabled: true,
+      traceChildRuns: false,
+      inputs: [],
+      outputs: [],
+    }
+    queueTableRows(schemaMock.customBlock, [
+      {
+        block,
+        workflowName: 'Lead workflow',
+        workspaceId: 'workspace-source',
+        workspaceName: 'Source workspace',
+      },
+    ])
+    loadDeployedWorkflowState.mockResolvedValue({ blocks: { start: { type: 'start' } } })
+    extractInputFieldsFromBlocks.mockReturnValue([])
+
+    const result = await listCustomBlocksWithInputs('org-1')
+
+    expect(result).toHaveLength(1)
+    expect(loadDeployedWorkflowState).toHaveBeenCalledWith('workflow-1', 'workspace-source')
+  })
+
+  it('bounds concurrent deployed-state hydration', async () => {
+    queueTableRows(
+      schemaMock.customBlock,
+      Array.from({ length: 11 }, (_, index) => ({
+        block: {
+          id: `custom-block-${index}`,
+          organizationId: 'org-1',
+          workflowId: `workflow-${index}`,
+          type: `custom_block_${index}`,
+          name: `Block ${index}`,
+          description: '',
+          iconUrl: null,
+          enabled: true,
+          traceChildRuns: false,
+          inputs: [],
+          outputs: [],
+        },
+        workflowName: `Workflow ${index}`,
+        workspaceId: 'workspace-source',
+        workspaceName: 'Source workspace',
+      }))
+    )
+    let active = 0
+    let maxActive = 0
+    loadDeployedWorkflowState.mockImplementation(async () => {
+      active++
+      maxActive = Math.max(maxActive, active)
+      await Promise.resolve()
+      active--
+      return { blocks: { start: { type: 'start' } } }
+    })
+    extractInputFieldsFromBlocks.mockReturnValue([])
+
+    const result = await listCustomBlocksWithInputs('org-1')
+
+    expect(result).toHaveLength(11)
+    expect(maxActive).toBe(10)
+  })
+})
+
 describe('reserved exposed-output names', () => {
  it('publishCustomBlock rejects an output named cost', async () => {
    await expect(
@@ -11,6 +11,7 @@ import { generateId, generateShortId } from '@sim/utils/id'
 import { and, eq, isNull, sql } from 'drizzle-orm'
 import { isOrganizationOnEnterprisePlan } from '@/lib/billing/core/subscription'
 import { isFeatureEnabled } from '@/lib/core/config/feature-flags'
+import { mapWithConcurrency } from '@/lib/core/utils/concurrency'
 import { extractInputFieldsFromBlocks, type WorkflowInputField } from '@/lib/workflows/input-format'
 import { loadDeployedWorkflowState } from '@/lib/workflows/persistence/utils'
 import { getWorkspaceWithOwner } from '@/lib/workspaces/permissions/utils'
@@ -18,6 +19,7 @@ import type { CustomBlockOutput, CustomBlockRow } from '@/blocks/custom/build-co
 import { CUSTOM_BLOCK_TYPE_PREFIX, isReservedOutputName } from '@/blocks/custom/build-config'

 const logger = createLogger('CustomBlocksOperations')
+const CUSTOM_BLOCK_HYDRATION_CONCURRENCY = 10

 /**
 * Resolve a workspace's organization ONLY when custom blocks are enabled for it —
@@ -82,9 +84,12 @@ export interface CustomBlockWithInputs {
 * expects) whenever the publisher edits after deploying. Returns `[]` if the
 * workflow has no active deployment.
 */
-async function deriveInputFields(workflowId: string): Promise<WorkflowInputField[]> {
+async function deriveInputFields(
+  workflowId: string,
+  workspaceId?: string
+): Promise<WorkflowInputField[]> {
  try {
-    const deployed = await loadDeployedWorkflowState(workflowId)
+    const deployed = await loadDeployedWorkflowState(workflowId, workspaceId)
    return extractInputFieldsFromBlocks(deployed.blocks)
  } catch {
    return []
@@ -220,7 +225,10 @@ async function hydrateCustomBlockRow(joined: {
    iconUrl: row.iconUrl,
    enabled: row.enabled,
    traceChildRuns: row.traceChildRuns,
-    inputFields: applyInputPlaceholders(row.inputs, await deriveInputFields(row.workflowId)),
+    inputFields: applyInputPlaceholders(
+      row.inputs,
+      await deriveInputFields(row.workflowId, workspaceId ?? undefined)
+    ),
    exposedOutputs: row.outputs ?? [],
  }
 }
@@ -241,7 +249,7 @@ export async function listCustomBlocksWithInputs(
    .leftJoin(workspace, eq(workspace.id, workflow.workspaceId))
    .where(eq(customBlock.organizationId, organizationId))

-  return Promise.all(rows.map(hydrateCustomBlockRow))
+  return mapWithConcurrency(rows, CUSTOM_BLOCK_HYDRATION_CONCURRENCY, hydrateCustomBlockRow)
 }

 /**
@@ -546,7 +554,10 @@ export async function publishCustomBlock(params: {
    iconUrl: iconUrl ?? null,
    enabled: true,
    traceChildRuns,
-    inputFields: applyInputPlaceholders(inputs ?? null, await deriveInputFields(workflowId)),
+    inputFields: applyInputPlaceholders(
+      inputs ?? null,
+      await deriveInputFields(workflowId, workspaceId)
+    ),
    exposedOutputs: exposedOutputs ?? [],
  }
 }
@@ -6,6 +6,7 @@ import { beforeEach, describe, expect, it, vi } from 'vitest'
 const mocks = vi.hoisted(() => ({
  listAccessible: vi.fn(),
  getDetail: vi.fn(),
+  getDetails: vi.fn(),
  loadContext: vi.fn(),
 }))

@@ -14,6 +15,7 @@ vi.mock('@/lib/workspaces/utils', () => ({
 }))
 vi.mock('@/lib/workspaces/public-queries', () => ({
  getPublicWorkspaceDetail: mocks.getDetail,
+  getPublicWorkspaceDetails: mocks.getDetails,
 }))
 vi.mock('@/lib/workspaces/application/workspace-context', () => ({
  loadActiveWorkspaceApplicationContext: mocks.loadContext,
@@ -41,6 +43,21 @@ describe('listPublicWorkspaces', () => {
      createdAt: new Date('2026-01-01T00:00:00Z'),
      updatedAt: new Date('2026-01-02T00:00:00Z'),
    }))
+    mocks.getDetails.mockImplementation(async (ids: string[]) => {
+      const details = new Map()
+      for (const id of [...ids].reverse()) {
+        details.set(id, {
+          id,
+          name: id,
+          color: '#33C482',
+          logoUrl: null,
+          memberCount: 1,
+          createdAt: new Date('2026-01-01T00:00:00Z'),
+          updatedAt: new Date('2026-01-02T00:00:00Z'),
+        })
+      }
+      return details
+    })
  })

  it('lists every accessible workspace where personal keys are enabled', async () => {
@@ -69,7 +86,55 @@ describe('listPublicWorkspaces', () => {

    expect(result.workspaces.map(({ id }) => id)).toEqual(['workspace-a'])
    expect(result.hasMore).toBe(true)
-    expect(mocks.getDetail).not.toHaveBeenCalledWith('workspace-disabled')
+    expect(mocks.getDetails).toHaveBeenCalledWith(['workspace-a'])
+    expect(mocks.getDetail).not.toHaveBeenCalled()
+  })
+
+  it('preserves the sorted page order independently of batch result order', async () => {
+    mocks.listAccessible.mockResolvedValue([
+      {
+        workspace: workspace('workspace-a', 'Alpha', true, 1),
+        permissionType: 'read',
+        viaOrgAdmin: false,
+      },
+      {
+        workspace: workspace('workspace-c', 'Charlie', true, 3),
+        permissionType: 'read',
+        viaOrgAdmin: false,
+      },
+      {
+        workspace: workspace('workspace-b', 'Beta', true, 2),
+        permissionType: 'read',
+        viaOrgAdmin: false,
+      },
+    ])
+
+    const result = await listPublicWorkspaces.execute({
+      principal: { kind: 'personal_api_key', userId: 'user-1', keyId: 'key-1' },
+      input: { sortBy: 'name', sortOrder: 'desc', limit: 2, offset: 0 },
+    })
+
+    expect(result.workspaces.map(({ id }) => id)).toEqual(['workspace-c', 'workspace-b'])
+    expect(result.hasMore).toBe(true)
+    expect(mocks.getDetails).toHaveBeenCalledWith(['workspace-c', 'workspace-b'])
+  })
+
+  it('fails when an accessible workspace disappears during batch hydration', async () => {
+    mocks.listAccessible.mockResolvedValue([
+      {
+        workspace: workspace('workspace-a', 'Alpha', true, 1),
+        permissionType: 'read',
+        viaOrgAdmin: false,
+      },
+    ])
+    mocks.getDetails.mockResolvedValue(new Map())
+
+    await expect(
+      listPublicWorkspaces.execute({
+        principal: { kind: 'personal_api_key', userId: 'user-1', keyId: 'key-1' },
+        input: { sortBy: 'name', sortOrder: 'asc', limit: 10, offset: 0 },
+      })
+    ).rejects.toThrow('Accessible workspace workspace-a disappeared during listing')
  })

  it('limits a workspace key to its bound active workspace', async () => {
@@ -9,6 +9,7 @@ import { workspaceOperations } from '@/lib/workspaces/application/operations'
 import { loadActiveWorkspaceApplicationContext } from '@/lib/workspaces/application/workspace-context'
 import {
  getPublicWorkspaceDetail,
+  getPublicWorkspaceDetails,
  type PublicWorkspaceDetail,
 } from '@/lib/workspaces/public-queries'
 import { listAccessibleWorkspaceRowsForUser } from '@/lib/workspaces/utils'
@@ -79,7 +80,12 @@ export const listPublicWorkspaces: OperationUseCase<
      .map(({ workspace }) => workspace)
      .sort((left, right) => compareWorkspaceRows(left, right, input.sortBy, input.sortOrder))
    const page = sorted.slice(input.offset, input.offset + input.limit)
-    const workspaces = await Promise.all(page.map(({ id }) => requirePublicWorkspaceDetail(id)))
+    const details = await getPublicWorkspaceDetails(page.map(({ id }) => id))
+    const workspaces = page.map(({ id }) => {
+      const workspace = details.get(id)
+      if (!workspace) throw new Error(`Accessible workspace ${id} disappeared during listing`)
+      return workspace
+    })

    return {
      workspaces,
@@ -3,7 +3,74 @@
 */
 import { dbChainMockFns, queueTableRows, resetDbChainMock, schemaMock } from '@sim/testing'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
-import { queryPublicWorkspaceMembers } from '@/lib/workspaces/public-queries'
+import {
+  getPublicWorkspaceDetails,
+  queryPublicWorkspaceMembers,
+} from '@/lib/workspaces/public-queries'
+
+describe('getPublicWorkspaceDetails', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetDbChainMock()
+  })
+
+  it('hydrates member counts for a bounded workspace batch', async () => {
+    queueTableRows(schemaMock.workspace, [
+      {
+        id: 'workspace-b',
+        name: 'Beta',
+        color: '#222222',
+        logoUrl: null,
+        organizationId: 'org-1',
+        createdAt: new Date('2026-01-02T00:00:00.000Z'),
+        updatedAt: new Date('2026-02-02T00:00:00.000Z'),
+      },
+      {
+        id: 'workspace-a',
+        name: 'Alpha',
+        color: '#111111',
+        logoUrl: null,
+        organizationId: null,
+        createdAt: new Date('2026-01-01T00:00:00.000Z'),
+        updatedAt: new Date('2026-02-01T00:00:00.000Z'),
+      },
+    ])
+    dbChainMockFns.execute.mockResolvedValueOnce([
+      { workspaceId: 'workspace-a', count: '2' },
+      { workspaceId: 'workspace-b', count: 4 },
+    ])
+
+    const details = await getPublicWorkspaceDetails(['workspace-a', 'workspace-b'])
+
+    expect(details.get('workspace-a')).toMatchObject({
+      id: 'workspace-a',
+      name: 'Alpha',
+      memberCount: 2,
+    })
+    expect(details.get('workspace-b')).toMatchObject({
+      id: 'workspace-b',
+      name: 'Beta',
+      memberCount: 4,
+    })
+    expect(dbChainMockFns.execute).toHaveBeenCalledTimes(1)
+    const countQuery = dbChainMockFns.execute.mock.calls[0][0] as {
+      values: Array<{
+        fragments: Array<{ strings: string[]; values: unknown[] }>
+      }>
+    }
+    const targetRows = countQuery.values[0].fragments
+    expect(
+      targetRows.every((fragment) => fragment.strings.join('?') === '(?::text, ?::text)')
+    ).toBe(true)
+    expect(targetRows.flatMap((fragment) => fragment.values)).toContain(null)
+  })
+
+  it('does not run a count query for an empty batch', async () => {
+    await expect(getPublicWorkspaceDetails([])).resolves.toEqual(new Map())
+    expect(dbChainMockFns.select).not.toHaveBeenCalled()
+    expect(dbChainMockFns.execute).not.toHaveBeenCalled()
+  })
+})

 describe('queryPublicWorkspaceMembers', () => {
  beforeEach(() => {
@@ -33,41 +33,65 @@ export interface WorkspaceMemberPage {
  nextEmail: string | null
 }

+interface WorkspaceMemberCountRow extends Record<string, unknown> {
+  workspaceId: string
+  count: number | string
+}
+
 async function countWorkspaceMembers(
-  workspaceId: string,
-  organizationId: string | null
-): Promise<number> {
+  workspaceRows: Array<{ id: string; organizationId: string | null }>
+): Promise<Map<string, number>> {
+  const counts = new Map<string, number>()
+  if (workspaceRows.length === 0) return counts
+
  const orgAdminRoles = sql.join(
    ORG_ADMIN_ROLES.map((role) => sql`${role}`),
    sql`, `
  )
-  const [row] = await db.execute<{ count: number | string }>(sql`
-    SELECT COUNT(*)::integer AS count
-    FROM (
+  const targets = sql.join(
+    workspaceRows.map(({ id, organizationId }) => sql`(${id}::text, ${organizationId}::text)`),
+    sql`, `
+  )
+  const rows = await db.execute<WorkspaceMemberCountRow>(sql`
+    WITH targets (workspace_id, organization_id) AS (VALUES ${targets})
+    SELECT
+      targets.workspace_id AS "workspaceId",
+      COUNT(effective_members.user_id)::integer AS count
+    FROM targets
+    LEFT JOIN LATERAL (
      SELECT ${permissions.userId} AS user_id
      FROM ${permissions}
      WHERE ${permissions.entityType} = 'workspace'
-        AND ${permissions.entityId} = ${workspaceId}
+        AND ${permissions.entityId} = targets.workspace_id
      UNION
      SELECT ${member.userId} AS user_id
      FROM ${member}
-      WHERE ${member.organizationId} = ${organizationId}
+      WHERE ${member.organizationId} = targets.organization_id
        AND ${member.role} IN (${orgAdminRoles})
-    ) AS effective_members
+    ) AS effective_members ON true
+    GROUP BY targets.workspace_id
  `)

-  const count = Number(row?.count)
-  if (!Number.isSafeInteger(count) || count < 0) {
-    throw new Error(`Invalid member count for workspace ${workspaceId}`)
+  for (const row of rows) {
+    const count = Number(row.count)
+    if (!Number.isSafeInteger(count) || count < 0) {
+      throw new Error(`Invalid member count for workspace ${row.workspaceId}`)
+    }
+    counts.set(row.workspaceId, count)
  }
-  return count
+
+  return counts
 }

 /** Public workspace metadata with all governance and billing identities omitted. */
-export async function getPublicWorkspaceDetail(
-  workspaceId: string
-): Promise<PublicWorkspaceDetail | null> {
-  const [row] = await db
+export async function getPublicWorkspaceDetails(
+  workspaceIds: string[]
+): Promise<Map<string, PublicWorkspaceDetail>> {
+  const details = new Map<string, PublicWorkspaceDetail>()
+  if (workspaceIds.length === 0) return details
+  const uniqueWorkspaceIds = [...new Set(workspaceIds)]
+
+  const rows = await db
    .select({
      id: workspace.id,
      name: workspace.name,
@@ -78,20 +102,33 @@ export async function getPublicWorkspaceDetail(
      updatedAt: workspace.updatedAt,
    })
    .from(workspace)
-    .where(and(eq(workspace.id, workspaceId), isNull(workspace.archivedAt)))
-    .limit(1)
+    .where(and(inArray(workspace.id, uniqueWorkspaceIds), isNull(workspace.archivedAt)))

-  if (!row) return null
-
-  return {
-    id: row.id,
-    name: row.name,
-    color: row.color,
-    logoUrl: row.logoUrl,
-    memberCount: await countWorkspaceMembers(workspaceId, row.organizationId),
-    createdAt: row.createdAt,
-    updatedAt: row.updatedAt,
+  const memberCounts = await countWorkspaceMembers(rows)
+  for (const row of rows) {
+    const memberCount = memberCounts.get(row.id)
+    if (memberCount === undefined) {
+      throw new Error(`Invalid member count for workspace ${row.id}`)
+    }
+    details.set(row.id, {
+      id: row.id,
+      name: row.name,
+      color: row.color,
+      logoUrl: row.logoUrl,
+      memberCount,
+      createdAt: row.createdAt,
+      updatedAt: row.updatedAt,
+    })
  }
+
+  return details
+}
+
+export async function getPublicWorkspaceDetail(
+  workspaceId: string
+): Promise<PublicWorkspaceDetail | null> {
+  const details = await getPublicWorkspaceDetails([workspaceId])
+  return details.get(workspaceId) ?? null
 }

 /**
@@ -0,0 +1,27 @@
+-- migration-safe: tune the append-heavy ledger's vacuum thresholds without changing row data or query semantics.
+ALTER TABLE "usage_log" SET (autovacuum_vacuum_insert_scale_factor = 0.01, autovacuum_analyze_scale_factor = 0.01);--> statement-breakpoint
+-- Concurrent index operations cannot run inside the migration runner's transaction.
+COMMIT;--> statement-breakpoint
+SET lock_timeout = 0;--> statement-breakpoint
+-- migration-safe: replay removes an invalid build created by this migration; concurrent operations preserve writes.
+DROP INDEX CONCURRENTLY IF EXISTS "doc_active_kb_token_count_idx";--> statement-breakpoint
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "doc_active_kb_token_count_idx" ON "document" USING btree ("knowledge_base_id","token_count") WHERE "document"."user_excluded" = false AND "document"."archived_at" IS NULL AND "document"."deleted_at" IS NULL;--> statement-breakpoint
+-- migration-safe: the existing connector_id index remains available until the ordered replacement is valid.
+DROP INDEX CONCURRENTLY IF EXISTS "kcsl_connector_started_at_idx";--> statement-breakpoint
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "kcsl_connector_started_at_idx" ON "knowledge_connector_sync_log" USING btree ("connector_id","started_at" DESC);--> statement-breakpoint
+-- migration-safe: replay removes an invalid build created by this migration; concurrent operations preserve writes.
+DROP INDEX CONCURRENTLY IF EXISTS "table_views_workspace_created_idx";--> statement-breakpoint
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "table_views_workspace_created_idx" ON "table_views" USING btree ("workspace_id","created_at","id");--> statement-breakpoint
+-- migration-safe: replay removes an invalid build created by this migration; concurrent operations preserve writes.
+DROP INDEX CONCURRENTLY IF EXISTS "workflow_active_workspace_sort_idx";--> statement-breakpoint
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "workflow_active_workspace_sort_idx" ON "workflow" USING btree ("workspace_id","sort_order","created_at","id") WHERE "workflow"."archived_at" IS NULL;--> statement-breakpoint
+-- migration-safe: each removed index duplicates a primary or unique index with the identical key definition.
+DROP INDEX CONCURRENTLY IF EXISTS "academy_certificate_number_idx";--> statement-breakpoint
+DROP INDEX CONCURRENTLY IF EXISTS "copilot_async_tool_calls_tool_call_id_idx";--> statement-breakpoint
+DROP INDEX CONCURRENTLY IF EXISTS "mothership_settings_workspace_id_idx";--> statement-breakpoint
+DROP INDEX CONCURRENTLY IF EXISTS "permissions_user_entity_idx";--> statement-breakpoint
+DROP INDEX CONCURRENTLY IF EXISTS "session_token_idx";--> statement-breakpoint
+DROP INDEX CONCURRENTLY IF EXISTS "workspace_file_key_idx";--> statement-breakpoint
+-- migration-safe: the valid ordered connector index covers the old index's complete key definition.
+DROP INDEX CONCURRENTLY IF EXISTS "kcsl_connector_id_idx";--> statement-breakpoint
+SET lock_timeout = '5s';
@@ -2115,6 +2115,13 @@
      "when": 1787432959648,
      "tag": "0302_fat_sentry",
      "breakpoints": true
+    },
+    {
+      "idx": 303,
+      "version": "7",
+      "when": 1787506599710,
+      "tag": "0303_colorful_gressill",
+      "breakpoints": true
    }
  ]
 }
@@ -78,7 +78,6 @@ export const session = pgTable(
  },
  (table) => ({
    userIdIdx: index('session_user_id_idx').on(table.userId),
-    tokenIdx: index('session_token_idx').on(table.token),
  })
 )

@@ -290,6 +289,9 @@ export const workflow = pgTable(
      .on(table.workspaceId, sql`coalesce(${table.folderId}, '')`, table.name)
      .where(sql`${table.archivedAt} IS NULL`),
    folderSortIdx: index('workflow_folder_sort_idx').on(table.folderId, table.sortOrder),
+    activeWorkspaceSortIdx: index('workflow_active_workspace_sort_idx')
+      .on(table.workspaceId, table.sortOrder, table.createdAt, table.id)
+      .where(sql`${table.archivedAt} IS NULL`),
    archivedAtIdx: index('workflow_archived_at_idx').on(table.archivedAt),
    workspaceArchivedAtPartialIdx: index('workflow_workspace_archived_partial_idx')
      .on(table.workspaceId, table.archivedAt)
@@ -1328,22 +1330,16 @@ export const skillMember = pgTable(
  })
 )

-export const mothershipSettings = pgTable(
-  'mothership_settings',
-  {
-    workspaceId: text('workspace_id')
-      .primaryKey()
-      .references(() => workspace.id, { onDelete: 'cascade' }),
-    mcpToolRefs: jsonb('mcp_tool_refs').notNull().default(sql`'[]'::jsonb`),
-    customToolRefs: jsonb('custom_tool_refs').notNull().default(sql`'[]'::jsonb`),
-    skillRefs: jsonb('skill_refs').notNull().default(sql`'[]'::jsonb`),
-    createdAt: timestamp('created_at').notNull().defaultNow(),
-    updatedAt: timestamp('updated_at').notNull().defaultNow(),
-  },
-  (table) => ({
-    workspaceIdIdx: index('mothership_settings_workspace_id_idx').on(table.workspaceId),
-  })
-)
+export const mothershipSettings = pgTable('mothership_settings', {
+  workspaceId: text('workspace_id')
+    .primaryKey()
+    .references(() => workspace.id, { onDelete: 'cascade' }),
+  mcpToolRefs: jsonb('mcp_tool_refs').notNull().default(sql`'[]'::jsonb`),
+  customToolRefs: jsonb('custom_tool_refs').notNull().default(sql`'[]'::jsonb`),
+  skillRefs: jsonb('skill_refs').notNull().default(sql`'[]'::jsonb`),
+  createdAt: timestamp('created_at').notNull().defaultNow(),
+  updatedAt: timestamp('updated_at').notNull().defaultNow(),
+})

 export const subscription = pgTable(
  'subscription',
@@ -2044,7 +2040,6 @@ export const workspaceFile = pgTable(
  },
  (table) => ({
    workspaceIdIdx: index('workspace_file_workspace_id_idx').on(table.workspaceId),
-    keyIdx: index('workspace_file_key_idx').on(table.key),
    deletedAtIdx: index('workspace_file_deleted_at_idx').on(table.deletedAt),
    workspaceDeletedAtPartialIdx: index('workspace_file_workspace_deleted_partial_idx')
      .on(table.workspaceId, table.deletedAt)
@@ -2406,13 +2401,6 @@ export const permissions = pgTable(
      table.permissionType
    ),

-    // User + specific entity queries - get user's permissions for specific entity
-    userEntityIdx: index('permissions_user_entity_idx').on(
-      table.userId,
-      table.entityType,
-      table.entityId
-    ),
-
    // Uniqueness constraint - prevent duplicate permission rows (one permission per user/entity)
    uniquePermissionConstraint: uniqueIndex('permissions_unique_constraint').on(
      table.userId,
@@ -2633,6 +2621,11 @@ export const document = pgTable(
      .where(sql`${table.deletedAt} IS NULL`),
    // Sync engine: load all active docs for a connector
    connectorIdIdx: index('doc_connector_id_idx').on(table.connectorId),
+    activeKnowledgeBaseTokenCountIdx: index('doc_active_kb_token_count_idx')
+      .on(table.knowledgeBaseId, table.tokenCount)
+      .where(
+        sql`${table.userExcluded} = false AND ${table.archivedAt} IS NULL AND ${table.deletedAt} IS NULL`
+      ),
    // KB file-access liveness: exact lookup by canonical storage key
    storageKeyIdx: index('doc_storage_key_idx')
      .on(table.storageKey)
@@ -3229,7 +3222,6 @@ export const copilotAsyncToolCalls = pgTable(
  (table) => ({
    runIdIdx: index('copilot_async_tool_calls_run_id_idx').on(table.runId),
    checkpointIdIdx: index('copilot_async_tool_calls_checkpoint_id_idx').on(table.checkpointId),
-    toolCallIdIdx: index('copilot_async_tool_calls_tool_call_id_idx').on(table.toolCallId),
    statusIdx: index('copilot_async_tool_calls_status_idx').on(table.status),
    runStatusIdx: index('copilot_async_tool_calls_run_status_idx').on(table.runId, table.status),
    toolCallUnique: uniqueIndex('copilot_async_tool_calls_tool_call_id_unique').on(
@@ -4426,7 +4418,10 @@ export const knowledgeConnectorSyncLog = pgTable(
    errorMessage: text('error_message'),
  },
  (table) => ({
-    connectorIdIdx: index('kcsl_connector_id_idx').on(table.connectorId),
+    connectorStartedAtIdx: index('kcsl_connector_started_at_idx').on(
+      table.connectorId,
+      sql`${table.startedAt} DESC`
+    ),
    /**
     * Serves the scheduler's five-minute sweep for orphaned `started` rows.
     *
@@ -4642,8 +4637,13 @@ export const tableViews = pgTable(
    updatedAt: timestamp('updated_at').notNull().defaultNow(),
  },
  (table) => ({
-    /** Covers the only read path: list a table's views in creation order. */
    tableCreatedIdx: index('table_views_table_created_idx').on(table.tableId, table.createdAt),
+    /** Covers workspace-scoped hydration without scanning every saved view. */
+    workspaceCreatedIdx: index('table_views_workspace_created_idx').on(
+      table.workspaceId,
+      table.createdAt,
+      table.id
+    ),
    /** At most one default view per table, enforced in the DB. */
    defaultViewUnique: uniqueIndex('table_views_table_default_unique')
      .on(table.tableId)
@@ -4913,7 +4913,6 @@ export const academyCertificate = pgTable(
      table.userId,
      table.courseId
    ),
-    certNumberIdx: index('academy_certificate_number_idx').on(table.certificateNumber),
    statusIdx: index('academy_certificate_status_idx').on(table.status),
  })
 )
@@ -1124,6 +1124,11 @@ export const schemaMock = {
    description: 'usageLog.description',
    metadata: 'usageLog.metadata',
    cost: 'usageLog.cost',
+    eventKey: 'usageLog.eventKey',
+    billingEntityType: 'usageLog.billingEntityType',
+    billingEntityId: 'usageLog.billingEntityId',
+    billingPeriodStart: 'usageLog.billingPeriodStart',
+    billingPeriodEnd: 'usageLog.billingPeriodEnd',
    workspaceId: 'usageLog.workspaceId',
    workflowId: 'usageLog.workflowId',
    executionId: 'usageLog.executionId',