fix(file-search): claim owners woken by a file write ahead of backfill reconciliation

apps/sim/lib/workspace-files/search/README.md

@@ -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.
+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 claim pass takes owners never dispatched first, so the owners a backfill page wakes enter the queue as if just dispatched: a page of reconciliation wake-ups rotates in by dispatch age behind the owners the file trigger woke for a write, instead of delaying them by several passes.
 
 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.

apps/sim/lib/workspace-files/search/dispatcher.integration.ts

@@ -29,6 +29,7 @@ import {
   FILE_SEARCH_BACKFILL_PAGE_SIZE,
   FILE_SEARCH_DISPATCH_HANDOFF_MS,
   FILE_SEARCH_INDEX_DISPATCH_WORKSPACES,
+  FILE_SEARCH_INDEX_MAX_OUTSTANDING,
   FILE_SEARCH_INDEX_STALE_DISPATCH_MS,
   FILE_SEARCH_RECONCILE_INTERVAL_MS,
 } from '@/lib/workspace-files/search/constants'
@@ -380,6 +381,84 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => {
     expect(row.active).toBe(100)
   })
 
+  /** Live files with ready revisions: owners a backfill page wakes with nothing to claim. */
+  async function seedIndexedOwners(prefix: string, count: number) {
+    await connection`INSERT INTO workspace_files (id, workspace_id, context, content_updated_at)
+      SELECT ${prefix} || lpad(n::text, 4, '0') || '-file', ${prefix} || lpad(n::text, 4, '0'),
+        'workspace', '2026-09-16'
+      FROM generate_series(1, ${count}) n`
+    await connection`INSERT INTO workspace_file_search_revision
+      (file_id, workspace_id, entity_type, entity_id, source_content_updated_at, status, updated_at)
+      SELECT id, workspace_id, 'workspace', workspace_id, content_updated_at, 'ready', '2026-09-16'
+      FROM workspace_files WHERE id LIKE ${`${prefix}%`}`
+  }
+
+  /**
+   * Writes a file through the real `workspace_file_search_mark_pending` trigger, attached only for
+   * this insert so the other fixtures keep seeding revisions and queue rows directly.
+   */
+  async function writeFile(workspaceId: string) {
+    await connection.begin(async (tx) => {
+      await tx`CREATE TRIGGER write_marks_pending
+        AFTER INSERT OR UPDATE OF content_updated_at, deleted_at, context, workspace_id
+        ON workspace_files FOR EACH ROW EXECUTE FUNCTION workspace_file_search_mark_pending()`
+      await tx`INSERT INTO workspace_files (id, workspace_id, context, content_updated_at)
+        VALUES (${`${workspaceId}-written`}, ${workspaceId}, 'workspace', '2026-09-16')`
+      await tx`DROP TRIGGER write_marks_pending ON workspace_files`
+    })
+  }
+
+  it('serves an owner woken by a file write ahead of a backfill page of reconciliation wake-ups', async () => {
+    const wakeUps = FILE_SEARCH_INDEX_DISPATCH_WORKSPACES * 2 + 50
+    await seedIndexedOwners('reconciled-', wakeUps)
+    const first = await prepareWorkspaceFileSearchDispatch()
+    expect(first.backfilledFiles).toBe(wakeUps)
+    expect(first.payloads).toEqual([])
+
+    await writeFile('written-workspace')
+    const next = await prepareWorkspaceFileSearchDispatch()
+    expect(next.payloads.map((payload) => payload.fileId)).toEqual(['written-workspace-written'])
+  })
+
+  it('serves a fresh write before work the backfill found, then the backfill work', async () => {
+    await seedQueue('workspace-active', 0, FILE_SEARCH_INDEX_MAX_OUTSTANDING)
+    await connection`DELETE FROM file_search_dispatch_queue WHERE entity_id = 'workspace-active'`
+    await connection`UPDATE workspace_file_search_backfill SET completed_at = NULL`
+    await connection`INSERT INTO workspace_files (id, workspace_id, context, content_updated_at)
+      VALUES ('backfill-found-file', 'backfill-found', 'workspace', '2026-09-16')`
+    /** No capacity: the page seeds and queues the unindexed file's owner but claims nothing. */
+    const seeded = await prepareWorkspaceFileSearchDispatch()
+    expect(seeded.backfilledFiles).toBeGreaterThan(0)
+    expect(seeded.payloads).toEqual([])
+
+    await writeFile('fresh-write')
+    const served: string[] = []
+    for (let pass = 0; pass < 2; pass++) {
+      /** Free exactly one slot, so each pass claims for one owner. */
+      await connection`UPDATE workspace_file_search_revision SET status = 'ready'
+        WHERE file_id = (SELECT file_id FROM workspace_file_search_revision
+          WHERE status = 'pending' AND dispatched_at IS NOT NULL ORDER BY file_id LIMIT 1)`
+      const result = await prepareWorkspaceFileSearchDispatch()
+      served.push(...result.payloads.map((payload) => payload.fileId))
+    }
+    expect(served).toEqual(['fresh-write-written', 'backfill-found-file'])
+  })
+
+  it('leaves an owner queued by a file write at the front when a backfill page passes it', async () => {
+    await seedQueue('workspace-active', 0, FILE_SEARCH_INDEX_MAX_OUTSTANDING - 1)
+    await connection`DELETE FROM file_search_dispatch_queue WHERE entity_id = 'workspace-active'`
+    /** Queued first 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'`
+    await writeFile('written-workspace')
+    await connection`UPDATE workspace_file_search_backfill SET completed_at = NULL`
+
+    const result = await prepareWorkspaceFileSearchDispatch()
+    expect(result.backfilledFiles).toBeGreaterThan(0)
+    expect(result.payloads.map((payload) => payload.fileId)).toEqual(['written-workspace-written'])
+  })
+
   it('walks every live workspace file exactly once across backfill pages', async () => {
     /** Deliberately not a whole number of pages, so the short final page ends the walk. */
     const files =

apps/sim/lib/workspace-files/search/dispatcher.ts

@@ -171,16 +171,30 @@ function queuedOwnerFilter(owners: readonly EditableFileOwner[]): SQL | undefine
   return or(...owners.map(queuedOwner))
 }
 
+/**
+ * Queues owners for the claim pass. A `reconciliation` wake-up, from the backfill walk, enters as
+ * if just dispatched: claims take owners never dispatched first, so marking it served keeps a page
+ * of owners with nothing new behind the owners the file trigger woke for a write, and its row still
+ * rotates in by dispatch age. An owner already queued keeps its place either way.
+ */
 async function enqueueOwners(
   tx: DbTransaction,
   owners: readonly EditableFileOwner[],
-  now: Date
+  now: Date,
+  { reconciliation = false }: { reconciliation?: boolean } = {}
 ): Promise<void> {
   const unique = [...new Map(owners.map((owner) => [JSON.stringify(owner), owner])).values()]
   if (unique.length === 0) return
   await tx
     .insert(fileSearchDispatchQueue)
-    .values(unique.map((owner) => ({ ...owner, enqueuedAt: now, updatedAt: now })))
+    .values(
+      unique.map((owner) => ({
+        ...owner,
+        enqueuedAt: now,
+        updatedAt: now,
+        lastDispatchedAt: reconciliation ? now : null,
+      }))
+    )
     .onConflictDoUpdate({
       target: [fileSearchDispatchQueue.entityType, fileSearchDispatchQueue.entityId],
       set: { updatedAt: now },
@@ -316,7 +330,7 @@ async function seedBackfillPage(tx: DbTransaction, now: Date): Promise<number> {
       .onConflictDoNothing()
   }
   /** Every owner the page passed is woken, indexed or not, so the pass reconciles its builds. */
-  await enqueueOwners(tx, rows.map(rowOwner), now)
+  await enqueueOwners(tx, rows.map(rowOwner), now, { reconciliation: true })
 
   const last = rows.at(-1)
   await tx