fix(session-query): reconcile concurrent live removals

packages/session-query/session-query/src/provider.ts

@@ -26,10 +26,15 @@ interface ProviderState {
   active: boolean
   chain: Promise<void>
   liveIds: Set<SessionId>
-  fullSync: Promise<void> | undefined
+  fullSync: FullSync | undefined
   liveSync: Map<SessionId, Promise<void>>
 }
 
+interface FullSync {
+  liveKey: string
+  promise: Promise<void>
+}
+
 /** Coordinates one selected provider against live and persisted corpus layers. */
 export class SessionProviderCoordinator {
   private readonly _configuredProviderId: string | undefined
@@ -159,7 +164,15 @@ export class SessionProviderCoordinator {
   }
 
   private _syncAll(state: ProviderState): Promise<void> {
-    if (state.fullSync !== undefined) return state.fullSync
+    // Capture the direct source before awaiting: only searches that observed
+    // the same live corpus may share an in-flight full synchronization.
+    const liveSessions = this._corpus().listLive()
+    const liveKey = JSON.stringify(liveSessions.map(session => this._snapshotLive(session)).map(snapshot => [
+      snapshot.session.header.id,
+      snapshot.fingerprint,
+      snapshot.session.persisted,
+    ]))
+    if (state.fullSync?.liveKey === liveKey) return state.fullSync.promise
     const promise = this._enqueue(state, async () => {
       /* v8 ignore next -- a provider can be disposed while queued behind an in-flight update */
       if (!state.active) return
@@ -169,12 +182,13 @@ export class SessionProviderCoordinator {
       } else {
         await this._syncPersisted(state, persistence)
       }
-      await this._replaceLiveCorpus(state, this._corpus().listLive())
+      await this._replaceLiveCorpus(state, liveSessions)
     })
-    state.fullSync = promise
+    const fullSync = { liveKey, promise }
+    state.fullSync = fullSync
     void promise.finally(() => {
       /* v8 ignore next -- a newer invalidation may already own the sync slot */
-      if (state.fullSync === promise) state.fullSync = undefined
+      if (state.fullSync === fullSync) state.fullSync = undefined
     }).catch(() => undefined)
     return promise
   }

packages/session-query/session-query/tests/session-query.spec.ts

@@ -467,6 +467,48 @@ describe('provider selection and synchronization', () => {
       .rejects.toThrow('search failed')
   })
 
+  it('reconciles a live removal observed while an older full sync is in flight', async () => {
+    const ctx = await liveContext()
+    const session = ctx.sessions.prepare(SessionId('removed-during-sync'))
+    const detach = ctx.sessions.enter(session)
+    ctx.sessions.announce(session)
+    session.append('user/message', { content: [{ type: 'text', text: 'stale live hit' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
+    const provider = new FakeProvider()
+    const replaceStarted = deferred()
+    const releaseReplace = deferred()
+    provider.replaceLive = async (snapshot) => {
+      replaceStarted.resolve()
+      await releaseReplace.promise
+      provider.live.set(snapshot.session.header.id, structuredClone(snapshot))
+    }
+    const searchLiveIds: SessionIdType[][] = []
+    provider.searchSessions = () => {
+      searchLiveIds.push([...provider.live.keys()])
+      const items: SessionSearchHit[] = []
+      for (const snapshot of provider.live.values()) {
+        const document = snapshot.documents[0]
+        if (document === undefined) continue
+        items.push({
+          ...structuredClone(snapshot.session),
+          bestMatch: { ...structuredClone(document), snippet: document.text },
+        })
+      }
+      return Promise.resolve({ providerId: provider.id, items })
+    }
+    ctx.sessionQuery.registerSearchProvider(provider)
+
+    const first = ctx.sessionQuery.searchSessions({ query: 'stale' })
+    await replaceStarted.promise
+    detach()
+    const second = ctx.sessionQuery.searchSessions({ query: 'stale' })
+    releaseReplace.resolve()
+
+    await first
+    await expect(second).resolves.toMatchObject({ items: [] })
+    expect(provider.removedLive).toContain(session.id)
+    expect(searchLiveIds.at(-1)).toEqual([])
+  })
+
   it('searches a persisted target after corpus reconciliation', async () => {
     const persisted = header('event-persisted', 1)
     TestPersistence.reset([{ meta: persisted, events: eventLog('persisted target') }])
@@ -528,6 +570,7 @@ describe('provider selection and synchronization', () => {
     await ctx.sessionQuery.searchSessions({ query: 'x' })
     expect(provider.persisted.get(persisted.id)?.documents[0]?.text).toBe('persisted')
     expect(provider.live.get(overlaid.id)?.documents[0]?.text).toBe('override')
+    expect(provider.live.get(overlaid.id)?.session).toMatchObject({ live: true, persisted: true })
     expect(provider.removedPersisted).toEqual([SessionId('stale')])
     expect(provider.activeHistory.at(-1)).toBe(true)
     const fingerprint = provider.persisted.get(persisted.id)?.fingerprint