@@ -1,14 +1,16 @@
 import { db } from '@sim/db'
 import { member, permissions, usageLog, workspace } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
-import { getPostgresCancellationReason } from '@sim/utils/errors'
 import { and, countDistinct, eq, inArray, notInArray, type SQL, sql } from 'drizzle-orm'
 import { unionAll } from 'drizzle-orm/pg-core'
-import { readLedgerBounded } from '@/lib/billing/core/ledger-read'
+import {
+  isLedgerReadOverBudget,
+  readLedgerBounded,
+  WORKFLOW_RUN_COUNT_BUDGET_MS,
+} from '@/lib/billing/core/ledger-read'
 import type { ResolvedUsagePeriod } from '@/lib/billing/core/reporting-period'
 import { buildUsageAnalyticsScope } from '@/lib/billing/core/usage-analytics'
 import type { BillingEntity } from '@/lib/billing/core/usage-log'
-import { DatabaseReadDeadlineError } from '@/lib/db/read-retry'
 
 /**
  * Ledger reads behind the admin dashboard: each billing entity's spend and workflow runs
@@ -26,7 +28,7 @@ import { DatabaseReadDeadlineError } from '@/lib/db/read-retry'
  * Widening both entity indexes with it would grow the hot billing sums' working set to
  * speed up a staff-only page. So only one organization's detail counts runs, in a single
  * scan that yields the total and every actor's share, under
- * {@link WORKFLOW_RUNS_BUDGET_MS}; past it the count is unknown (`null`) and the page
+ * {@link WORKFLOW_RUN_COUNT_BUDGET_MS}; past it the count is unknown (`null`) and the page
  * still loads.
  *
  * They stay on the primary, where the hot billing sums keep those index pages cached. On
@@ -54,12 +56,6 @@ interface DashboardWorkflowRuns {
   formerActors: number
 }
 
-/**
- * How long one detail page may spend counting workflow runs. Well inside the admin
- * proxy's request timeout, so a payer too large to count still gets its page.
- */
-const WORKFLOW_RUNS_BUDGET_MS = 5_000
-
 const WORKFLOW_SOURCE = sql`${usageLog.source} = 'workflow'`
 
 export function organizationWindow(
@@ -164,7 +160,7 @@ function isFormerActor(organizationId: string): SQL {
 /**
  * An organization's workflow runs over its period, in total, per acting user and across its
  * former actors, from one scan. `null` when the count does not finish within
- * {@link WORKFLOW_RUNS_BUDGET_MS}.
+ * {@link WORKFLOW_RUN_COUNT_BUDGET_MS}.
  */
 export async function readDashboardOrganizationWorkflowRuns(
   organizationId: string,
@@ -193,16 +189,13 @@ export async function readDashboardOrganizationWorkflowRuns(
           .from(workflowRows)
           .groupBy(sql`grouping sets ((${workflowRows.userId}), (${workflowRows.formerActor}), ())`)
       },
-      WORKFLOW_RUNS_BUDGET_MS
+      WORKFLOW_RUN_COUNT_BUDGET_MS
     )
   } catch (error) {
-    const overBudget =
-      error instanceof DatabaseReadDeadlineError ||
-      getPostgresCancellationReason(error) === 'statement_timeout'
-    if (!overBudget) throw error
+    if (!isLedgerReadOverBudget(error)) throw error
     logger.warn('Workflow run count exceeded its budget', {
       organizationId,
-      budgetMs: WORKFLOW_RUNS_BUDGET_MS,
+      budgetMs: WORKFLOW_RUN_COUNT_BUDGET_MS,
     })
     return null
   }
 
@@ -1,3 +1,4 @@
+import { getPostgresCancellationReason } from '@sim/utils/errors'
 import { sql } from 'drizzle-orm'
 import { USAGE_LEDGER_STATEMENT_TIMEOUT_MS } from '@/lib/billing/constants'
 import { DatabaseReadDeadlineError, withDatabaseReadRetry } from '@/lib/db/read-retry'
@@ -36,3 +37,17 @@ export function readLedgerBounded<T>(
     { label: 'usage ledger', maxAttempts: 2, maxElapsedMs: budgetMs }
   )
 }
+
+/**
+ * How long a staff-only workflow run count may take before it is reported as unknown. Well inside
+ * the admin proxy's request timeout, so a payer too large to count still gets its page.
+ */
+export const WORKFLOW_RUN_COUNT_BUDGET_MS = 5_000
+
+/** Whether a {@link readLedgerBounded} read failed because it ran out of its budget. */
+export function isLedgerReadOverBudget(error: unknown): boolean {
+  return (
+    error instanceof DatabaseReadDeadlineError ||
+    getPostgresCancellationReason(error) === 'statement_timeout'
+  )
+}
@@ -32,6 +32,7 @@ import {
   CumulativeUsagePeriodClosedError,
   getBillingPeriodUsageCost,
   getBillingPeriodUsageCostByUser,
+  getBillingPeriodWorkflowRunCountWithinBudget,
   getStampedPeriodRangeUsageCostByUser,
   type RecordCumulativeUsageParams,
   recordCumulativeUsage,
@@ -375,6 +376,38 @@ describe('Cumulative billing with PostgreSQL', () => {
     )
   })
 
+  it('reports a workflow run count as unknown once it outlasts the preview budget', async () => {
+    const entity = { type: 'organization' as const, id: 'org-runs' }
+    const period = {
+      start: new Date('2026-08-01T00:00:00.000Z'),
+      end: new Date('2026-09-01T00:00:00.000Z'),
+      source: 'reporting' as const,
+    }
+    await connection`
+      insert into usage_log (id, user_id, category, source, description, cost,
+        billing_entity_type, billing_entity_id, execution_id, created_at)
+      values ('r1', 'u1', 'model', 'workflow', 'run', 1, 'organization', 'org-runs', 'e1', '2026-08-02'),
+             ('r2', 'u1', 'tool', 'workflow', 'run', 1, 'organization', 'org-runs', 'e1', '2026-08-02'),
+             ('r3', 'u1', 'model', 'workflow', 'run', 1, 'organization', 'org-runs', 'e2', '2026-08-03')
+    `
+    expect(await getBillingPeriodWorkflowRunCountWithinBudget(entity, period)).toBe(2)
+
+    const release = deferred()
+    const locked = deferred()
+    const holder = connection.begin(async (tx) => {
+      await tx.unsafe('LOCK TABLE usage_log IN ACCESS EXCLUSIVE MODE')
+      locked.resolve()
+      await release.promise
+    })
+    try {
+      await locked.promise
+      expect(await getBillingPeriodWorkflowRunCountWithinBudget(entity, period)).toBeNull()
+    } finally {
+      release.resolve()
+      await holder
+    }
+  }, 20_000)
+
   describe('a request that outlives its billing period', () => {
     // Past periods: the old period's row is written under the subscription lock only once
     // that period has ended.
 
@@ -18,7 +18,11 @@ import {
   timestampKey,
 } from '@/lib/api/list-query'
 import { defaultBillingPeriod } from '@/lib/billing/core/billing-period'
-import { readLedgerBounded } from '@/lib/billing/core/ledger-read'
+import {
+  isLedgerReadOverBudget,
+  readLedgerBounded,
+  WORKFLOW_RUN_COUNT_BUDGET_MS,
+} from '@/lib/billing/core/ledger-read'
 import { getHighestPrioritySubscription } from '@/lib/billing/core/plan'
 import {
   resolveSubscriptionUsagePeriod,
@@ -269,44 +273,74 @@ export async function getBillingPeriodUsageCost(
 export async function getBillingPeriodWorkflowRunCount(
   billingEntity: BillingEntity,
   billingPeriod: UsageQueryPeriod,
-  executor: DbClient = db
+  executor: DbClient = db,
+  budgetMs?: number
 ): Promise<number> {
-  const [row] = await readLedgerBounded(executor, (tx) =>
-    tx
-      .select({
-        /**
-         * The exclusion goes through `notInArray`, not `<> ALL(${array})`. Interpolating
-         * a JavaScript array into a `sql` template emits parenthesized scalar binds —
-         * `ALL(($1))` — which Postgres rejects outright with "op ANY/ALL (array)
-         * requires array on right side". Unit tests cannot catch it, because `@sim/db`
-         * is mocked and no statement is ever rendered.
-         */
-        workflowRuns:
-          sql<number>`COUNT(DISTINCT ${usageLog.executionId}) FILTER (WHERE ${usageLog.source} = 'workflow' AND ${notInArray(usageLog.category, [...UNBILLED_USAGE_CATEGORIES])})`.mapWith(
-            Number
-          ),
-      })
-      .from(usageLog)
-      .where(
-        and(
-          eq(usageLog.billingEntityType, billingEntity.type),
-          eq(usageLog.billingEntityId, billingEntity.id),
-          ...(billingPeriod.source === 'reporting'
-            ? [
-                gte(usageLog.createdAt, billingPeriod.start),
-                lt(usageLog.createdAt, billingPeriod.end),
-              ]
-            : [
-                eq(usageLog.billingPeriodStart, billingPeriod.start),
-                eq(usageLog.billingPeriodEnd, billingPeriod.end),
-              ])
-        )
-      )
+  const [row] = await readLedgerBounded(
+    executor,
+    (tx) =>
+      tx
+        .select({
+          /**
+           * The exclusion goes through `notInArray`, not `<> ALL(${array})`. Interpolating
+           * a JavaScript array into a `sql` template emits parenthesized scalar binds —
+           * `ALL(($1))` — which Postgres rejects outright with "op ANY/ALL (array)
+           * requires array on right side". Unit tests cannot catch it, because `@sim/db`
+           * is mocked and no statement is ever rendered.
+           */
+          workflowRuns:
+            sql<number>`COUNT(DISTINCT ${usageLog.executionId}) FILTER (WHERE ${usageLog.source} = 'workflow' AND ${notInArray(usageLog.category, [...UNBILLED_USAGE_CATEGORIES])})`.mapWith(
+              Number
+            ),
+        })
+        .from(usageLog)
+        .where(
+          and(
+            eq(usageLog.billingEntityType, billingEntity.type),
+            eq(usageLog.billingEntityId, billingEntity.id),
+            ...(billingPeriod.source === 'reporting'
+              ? [
+                  gte(usageLog.createdAt, billingPeriod.start),
+                  lt(usageLog.createdAt, billingPeriod.end),
+                ]
+              : [
+                  eq(usageLog.billingPeriodStart, billingPeriod.start),
+                  eq(usageLog.billingPeriodEnd, billingPeriod.end),
+                ])
+          )
+        ),
+    budgetMs
   )
 
   return row?.workflowRuns ?? 0
 }
 
+/**
+ * {@link getBillingPeriodWorkflowRunCount} for a staff-only preview that can do without it:
+ * `null` when the count does not finish within {@link WORKFLOW_RUN_COUNT_BUDGET_MS}, so a payer
+ * too large to count still gets its preview inside the admin proxy's request timeout.
+ */
+export async function getBillingPeriodWorkflowRunCountWithinBudget(
+  billingEntity: BillingEntity,
+  billingPeriod: UsageQueryPeriod
+): Promise<number | null> {
+  try {
+    return await getBillingPeriodWorkflowRunCount(
+      billingEntity,
+      billingPeriod,
+      db,
+      WORKFLOW_RUN_COUNT_BUDGET_MS
+    )
+  } catch (error) {
+    if (!isLedgerReadOverBudget(error)) throw error
+    logger.warn('Billing period workflow run count exceeded its budget', {
+      billingEntityType: billingEntity.type,
+      budgetMs: WORKFLOW_RUN_COUNT_BUDGET_MS,
+    })
+    return null
+  }
+}
+
 /**
  * Period total plus the portion attributable to `source`, in a single scan.
  *
 
@@ -37,7 +37,7 @@ import { getBillingConcurrencyLimit } from '@/lib/billing/concurrency-limits'
 import { resolveEnterpriseReportingPeriod } from '@/lib/billing/core/reporting-period'
 import {
   getBillingPeriodUsageCost,
-  getBillingPeriodWorkflowRunCount,
+  getBillingPeriodWorkflowRunCountWithinBudget,
 } from '@/lib/billing/core/usage-log'
 import { creditsToDollars, dollarsToCredits } from '@/lib/billing/credits/conversion'
 import {
@@ -462,7 +462,8 @@ export interface EnterpriseIssuancePreflight {
       limitDollars: number
       usedCredits: number
       limitCredits: number
-      workflowRuns: number
+      /** `null` when the count did not finish within its budget. */
+      workflowRuns: number | null
     }
     invoiceAmountUsd: number
     configuredUsageLimitDollars: number
@@ -865,7 +866,7 @@ export async function getEnterpriseIssuancePreflight({
             { type: 'organization', id: membership.organizationId },
             reportingPeriod
           ),
-          getBillingPeriodWorkflowRunCount(
+          getBillingPeriodWorkflowRunCountWithinBudget(
             { type: 'organization', id: membership.organizationId },
             reportingPeriod
           ),
 
@@ -75,9 +75,9 @@ export function getBaseUrl(): string {
  * Whether this process is a Trigger.dev worker rather than the app container.
  *
  * `TRIGGER_SECRET_KEY` and `TRIGGER_DEV_ENABLED` are both present on the app too
- * — it needs them to dispatch — so neither discriminates. `trigger.config.ts`
- * syncs exactly one worker-only marker, `DB_APP_NAME='sim-trigger'`, which is
- * what this reads.
+ * — it needs them to dispatch — so neither discriminates. Every Trigger.dev
+ * deploy sets `DB_APP_NAME='sim-trigger'` (`TRIGGER_CONSTANT_VARS` in
+ * `@sim/runtime-secrets/trigger-env`), which the app never has, so this reads it.
  */
 function isTriggerWorkerRuntime(): boolean {
   return getEnv('DB_APP_NAME') === 'sim-trigger'
 
@@ -71,6 +71,6 @@ The design uses PostgreSQL's documented [TOAST behavior](https://www.postgresql.
 
 Migration `0426_file_search_owner_scope.sql` adds canonical workspace/Project ownership to the existing build, revision, and chunk pipeline. Legacy workspace inserts are translated by triggers, and the existing workspace trigram index remains usable. Project rows use a separate owner-scoped trigram index on the same chunk table. Organizations and personal users are not enabled search owners.
 
-The new `file_search_dispatch_queue` replaces the workspace-keyed scheduler because older workers cannot decode a Project without a workspace. During expansion, workspace mutations enqueue both queues, while the new dispatcher reads the owner queue. Both deployments share the dispatcher lock, revision claim token, build lease, and publication fence. Drain old app dispatchers and Trigger workers/retries before enabling Project dispatch. Retire the legacy queue and its dual enqueue in a later contract deploy; they are not a second indexing pipeline. The `file-search-owners-v3` cursor performs bounded owner/file keyset backfill and hourly reconciliation without rewriting extracted text in the migration. Claims take owners woken since their last dispatch first (`wokenSinceDispatch` in `dispatcher.ts`); backfill wake-ups enter as if just dispatched and never touch an owner already queued, so a page of them cannot delay a file write by several passes.
+The new `file_search_dispatch_queue` replaces the workspace-keyed scheduler because older workers cannot decode a Project without a workspace. During expansion, workspace mutations enqueue both queues, while the new dispatcher reads the owner queue. Both deployments share the dispatcher lock, revision claim token, build lease, and publication fence. Drain old app dispatchers and Trigger workers/retries before enabling Project dispatch. Retire the legacy queue and its dual enqueue in a later contract deploy; they are not a second indexing pipeline. The `file-search-owners-v3` cursor performs bounded owner/file keyset backfill and hourly reconciliation without rewriting extracted text in the migration. Each pass selects owners woken since their last dispatch that still have a free slot first (`wokenSinceDispatch` in `dispatcher.ts`), then the rest by dispatch age, and fills its capacity owner by owner in that order. Backfill wake-ups enter as if just dispatched and never touch an owner already queued, so a page of them rotates in behind owners woken for work.
 
 Generated document indexing never executes source. A cached artifact must match its owner and current input manifest. Each published build stores at most 500 input identities. Queries exclude a build when any input is changed, archived, deleted, or outside the owner, and report it as pending coverage. Input mutations wake their owner queue; the dispatcher retires stale manifests in bounded batches before rebuilding. Expired build cleanup also removes its dependency records. A cache miss indexes the stored generation source as text until an authorized rendering path creates a current artifact.
@@ -409,10 +409,14 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => {
     })
   }
 
-  /** Fills global capacity with claims of an unqueued owner, leaving `slots` free. */
+  /**
+   * Fills global capacity with claims of an owner that is neither queued nor walked by the
+   * backfill (its files are deleted), leaving `slots` free.
+   */
   async function leaveSlots(slots: number) {
     await seedQueue('workspace-active', 0, FILE_SEARCH_INDEX_MAX_OUTSTANDING - slots)
     await connection`DELETE FROM file_search_dispatch_queue WHERE entity_id = 'workspace-active'`
+    await connection`UPDATE workspace_files SET deleted_at = now() WHERE workspace_id = 'workspace-active'`
   }
 
   const wakeUps = FILE_SEARCH_INDEX_DISPATCH_WORKSPACES * 2 + 50
@@ -469,6 +473,40 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => {
     expect(result.payloads.map((payload) => payload.fileId)).toEqual(['waiting-workspace-000001'])
   })
 
+  it('fills the free capacity in selection order, so a woken owner is not outbid by an older backlog', async () => {
+    await leaveSlots(2)
+    // Served ten minutes ago and not woken since, with older pending files than the fresh write.
+    await seedQueue('backlog-workspace', 3)
+    await connection`UPDATE file_search_dispatch_queue SET last_dispatched_at = now() - interval '10 minutes',
+      updated_at = now() - interval '10 minutes' WHERE entity_id = 'backlog-workspace'`
+    await writeFile('fresh-workspace')
+
+    const result = await prepareWorkspaceFileSearchDispatch()
+    expect(result.payloads.map((payload) => payload.fileId).sort()).toEqual([
+      'backlog-workspace-000001',
+      'fresh-workspace-written',
+    ])
+  })
+
+  it('keeps a reaped claim woken when the backfill queues its owner in the same pass', async () => {
+    await leaveSlots(1)
+    // Its only claim was abandoned before its run was enqueued; nothing has queued its owner yet.
+    await seedQueue('reaped-workspace', 0, 1)
+    await connection`DELETE FROM file_search_dispatch_queue WHERE entity_id = 'reaped-workspace'`
+    await connection`UPDATE workspace_file_search_revision
+      SET handoff_expires_at = clock_timestamp() - interval '1 millisecond'
+      WHERE file_id = 'reaped-workspace-000001'`
+    // Served an hour ago and not woken since: it wins the single slot only if the reap is lost.
+    await seedQueue('waiting-workspace', 1)
+    await connection`UPDATE file_search_dispatch_queue SET last_dispatched_at = now() - interval '1 hour',
+      updated_at = now() - interval '1 hour' WHERE entity_id = 'waiting-workspace'`
+    await connection`UPDATE workspace_file_search_backfill SET completed_at = NULL`
+
+    const result = await prepareWorkspaceFileSearchDispatch()
+    expect(result.reapedClaims).toBe(1)
+    expect(result.payloads.map((payload) => payload.fileId)).toEqual(['reaped-workspace-000001'])
+  })
+
   it('serves a fresh write before work the backfill found, then the backfill work', async () => {
     await leaveSlots(0)
     await connection`UPDATE workspace_file_search_backfill SET completed_at = NULL`
@@ -494,7 +532,7 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => {
 
   it('leaves an owner queued by a file write at the front when a backfill page passes it', async () => {
     await leaveSlots(1)
-    // Queued first and served an hour ago: it wins the single slot only if the write is demoted.
+    // Woken and served an hour ago: it wins the single slot only if the write is demoted.
     await seedQueue('earlier-served', 1)
     await connection`UPDATE file_search_dispatch_queue
       SET last_dispatched_at = now() - interval '1 hour' WHERE entity_id = 'earlier-served'`