@@ -26,6 +26,7 @@ import {
   memberFailureBackoffMs,
   memberNextAttemptAt,
   nextMemberSyncTime,
+  persistedDocumentsByObserver,
   shouldListFully,
 } from '@/lib/knowledge/connectors/member-sync-engine'
 import {
@@ -272,6 +273,26 @@ describe('member sync engine decisions', () => {
       expect(second.retainedBytes).toBeGreaterThan(first.retainedBytes)
     })
 
+    it('grants a persisted batch to every member who listed each document, as it lands', () => {
+      const union = new Map()
+      admitMemberListing(union, 'm-1', [doc('a'), doc('b')], 'c-1', 0)
+      admitMemberListing(union, 'm-2', [doc('a'), doc('c')], 'c-1', 0)
+
+      const byMember = persistedDocumentsByObserver(
+        [
+          { externalId: 'a', documentId: 'd-a' },
+          { externalId: 'b', documentId: 'd-b' },
+          { externalId: 'zzz', documentId: 'd-z' },
+        ],
+        union
+      )
+
+      expect([...byMember.entries()]).toEqual([
+        ['m-1', ['d-a', 'd-b']],
+        ['m-2', ['d-a']],
+      ])
+    })
+
     it('counts a member once per external id even when their listing repeats it', () => {
       const union = new Map()
       const admitted = admitMemberListing(union, 'm-1', [doc('a'), doc('a')], 'c-1', 0)
 
@@ -69,6 +69,7 @@ import {
   classifySuspectListing,
   createSyncRunState,
   loadOwnedCorpus,
+  type PersistedDocument,
   processDocOps,
   RETRY_WINDOW_DAYS,
   runChangeFeedPass,
@@ -169,6 +170,27 @@ interface MemberListingOutcome {
   changeCursor: string | null | undefined
 }
 
+/**
+ * The documents a batch wrote, grouped by each member who listed them, so a
+ * grant can be recorded per observer the moment the row exists. A document
+ * nobody in the union listed (it cannot happen for a persisted one, but the
+ * map is the source of truth) grants nothing.
+ */
+export function persistedDocumentsByObserver(
+  persisted: readonly PersistedDocument[],
+  union: ReadonlyMap<string, Pick<UnionEntry, 'observers'>>
+): Map<string, string[]> {
+  const byMember = new Map<string, string[]>()
+  for (const { externalId, documentId } of persisted) {
+    for (const memberId of union.get(externalId)?.observers ?? []) {
+      const documentIds = byMember.get(memberId)
+      if (documentIds) documentIds.push(documentId)
+      else byMember.set(memberId, [documentId])
+    }
+  }
+  return byMember
+}
+
 interface UnionEntry {
   document: ExternalDocument
   /** Member ids whose listings returned the document, in listing order. */
@@ -1550,6 +1572,33 @@ export async function executeMemberSync(
       },
       lease: run.lease,
       documentAccess: 'members',
+      /**
+       * Grants surface as each batch lands: every member who listed a document
+       * observes it the moment its row exists, and its ACL is materialised in
+       * the same lease-proved transaction. The listing's own pass below records
+       * the same observations again, idempotently, and is still what decides
+       * removals; this only brings the additions forward from the end of the
+       * run to the moment they are indexed.
+       */
+      onBatchPersisted: async (persisted) => {
+        const byMember = persistedDocumentsByObserver(persisted, union)
+        if (byMember.size === 0) return
+        await withMemberLease(run, async (tx) => {
+          for (const [memberId, documentIds] of byMember) {
+            result.observationsAdded += await recordMemberObservations(
+              tx,
+              memberId,
+              documentIds,
+              runId
+            )
+          }
+          await materializeDocumentAcls(
+            connectorId,
+            persisted.map(({ documentId }) => documentId),
+            tx
+          )
+        })
+      },
     })
 
     const documentIdByExternalId = await loadDocumentIdsByExternalId(connectorId)
 
@@ -1496,6 +1496,12 @@ export interface ProcessDocOpsInput {
   state: SyncRunState
   hydration: DocOpHydration
   lease: Pick<SyncRunLease, 'beatIfDue' | 'beatLive' | 'stillHeld'>
+  /**
+   * Runs after each batch's rows are written, with the documents that landed.
+   * A members-mode run grants access here, batch by batch, so a long first
+   * crawl becomes searchable as it goes rather than all at once at the end.
+   */
+  onBatchPersisted?: (persisted: readonly PersistedDocument[]) => Promise<void>
   /** Who may read the documents this pass writes. */
   documentAccess: SyncDocumentAccess
 }
@@ -1507,6 +1513,12 @@ export interface ProcessDocOpsInput {
  * run so the connector backs off, and a lost lease, which ends it so no
  * further write lands beside the replacement run's.
  */
+/** A document a batch wrote, by the id the source knows it by and the id the row has. */
+export interface PersistedDocument {
+  externalId: string
+  documentId: string
+}
+
 export async function processDocOps(input: ProcessDocOpsInput): Promise<void> {
   const {
     connectorId,
@@ -1744,10 +1756,15 @@ export async function processDocOps(input: ProcessDocOpsInput): Promise<void> {
     if (leaseLost) throw leaseLost.reason
 
     const batchDocs: DocumentData[] = []
+    const persisted: PersistedDocument[] = []
     for (let j = 0; j < settled.length; j++) {
       const outcome = settled[j]
       if (outcome.status === 'fulfilled') {
         batchDocs.push(outcome.value)
+        persisted.push({
+          externalId: batch[j].extDoc.externalId,
+          documentId: outcome.value.documentId,
+        })
         if (batch[j].type === 'add') result.docsAdded++
         else result.docsUpdated++
       } else {
@@ -1761,6 +1778,8 @@ export async function processDocOps(input: ProcessDocOpsInput): Promise<void> {
       }
     }
 
+    if (persisted.length > 0) await input.onBatchPersisted?.(persisted)
+
     if (batchDocs.length > 0) {
       result.processingDispatch.requested += batchDocs.length
       try {