fix(session-query): harden provider operation lifecycle

docs/cordis-catalog/services.md

@@ -179,7 +179,7 @@ async listEvents(sessionId: SessionId): Promise<SessionEventRecord[]>
 async readEvent(request: SessionEventReadRequest): Promise<SessionEventWindow>
 async traceSession(sessionId: SessionId): Promise<SessionLineageTrace>
 async traceEvent(sessionId: SessionId, seq: number): Promise<SessionEventTrace>
-registerSearchProvider(provider: SessionSearchProvider): () => void
+registerSearchProvider(provider: SessionSearchProvider): () => Promise<void>
 registerEventTextExtractor<K extends SessionEventType>( type: K, extractor: SessionEventTextExtractor<K>, ): () => void
 registerContentTextExtractor<K extends ContentBlockType>( type: K, extractor: SessionContentTextExtractor<K>, ): () => void
 searchSessions( request: SessionSearchRequest, exec?: SessionQueryExecContext, ): Promise<SessionSearchPage<SessionSearchHit>>

packages/cordis/tool-cordis/src/api-catalog.ts

@@ -144,7 +144,7 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [
       'async readEvent(request: SessionEventReadRequest): Promise<SessionEventWindow>',
       'async traceSession(sessionId: SessionId): Promise<SessionLineageTrace>',
       'async traceEvent(sessionId: SessionId, seq: number): Promise<SessionEventTrace>',
-      'registerSearchProvider(provider: SessionSearchProvider): () => void',
+      'registerSearchProvider(provider: SessionSearchProvider): () => Promise<void>',
       'registerEventTextExtractor<K extends SessionEventType>( type: K, extractor: SessionEventTextExtractor<K>, ): () => void',
       'registerContentTextExtractor<K extends ContentBlockType>( type: K, extractor: SessionContentTextExtractor<K>, ): () => void',
       'searchSessions( request: SessionSearchRequest, exec?: SessionQueryExecContext, ): Promise<SessionSearchPage<SessionSearchHit>>',

packages/session-query/session-query/README.md

@@ -22,7 +22,7 @@ Session filters cover id, exact cwd, inclusive creation time, parent id/root, an
 
 ## Full-text providers
 
-`registerSearchProvider(provider)` is effect-scoped and ids are unique. Without `searchProvider`, exactly one locally available provider must be registered; explicit selection fails loudly when the named provider is missing or unavailable. Search pages default to 20 hits and reject limits above 100; a provider returning more hits than the normalized request limit fails with a typed provider error rather than silently dropping cursor-addressable results. Provider scores never cross the public API: event hits carry a plain snippet, while each session hit carries exactly one best matching event.
+`registerSearchProvider(provider)` is effect-scoped and ids are unique. Its async disposer removes the provider from selection immediately, lets already accepted transactions finish, and settles after they drain. Without `searchProvider`, exactly one locally available provider must be registered; explicit selection fails loudly when the named provider is missing or unavailable. Search pages default to 20 hits and reject limits above 100; a provider returning more hits than the normalized request limit fails with a typed provider error rather than silently dropping cursor-addressable results. Provider scores never cross the public API: event hits carry a plain snippet, while each session hit carries exactly one best matching event.
 
 The service feeds providers two independent layers: a durable persisted base (`persistedInventory`, `replacePersisted`, `removePersisted`, `setPersistedActive`) and an ephemeral live override (`replaceLive`, `removeLive`). A search waits for the relevant source state observed before its call: the whole corpus for session search, only the target for a live event search. Failed derived updates do not fail session writes; affected searches receive `SESSION_QUERY_INDEX_FAILED`, and a later search retries the dirty state. `AbortSignal` lets a caller stop waiting and is also passed to provider search.
 

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

@@ -151,9 +151,9 @@ export class SessionQueryService extends Service {
   /**
    * Register one full-text provider with effect-scoped disposal.
    * @param provider - provider and synchronization implementation.
-   * @returns disposer that unregisters the provider.
+   * @returns async disposer that immediately unregisters selection and awaits accepted provider work.
    */
-  registerSearchProvider(provider: SessionSearchProvider): () => void {
+  registerSearchProvider(provider: SessionSearchProvider): () => Promise<void> {
     return this._providers.register(this.ctx, provider)
   }
 

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

@@ -23,7 +23,6 @@ import { filterEventResults, filterSessionResults } from './filters.ts'
 
 interface ProviderState {
   provider: SessionSearchProvider
-  active: boolean
   chain: Promise<void>
   liveIds: Set<SessionId>
 }
@@ -49,26 +48,25 @@ export class SessionProviderCoordinator {
    * Register one effect-scoped provider.
    * @param ctx - contributing caller context.
    * @param provider - provider implementation.
-   * @returns disposer for the registration.
+   * @returns async disposer that deselects immediately and drains accepted work.
    */
-  register(ctx: Context, provider: SessionSearchProvider): () => void {
+  register(ctx: Context, provider: SessionSearchProvider): () => Promise<void> {
     if (this._providers.has(provider.id)) {
       throw new SessionQueryError(`a session-query provider with id "${provider.id}" is already registered`, 'SESSION_QUERY_DUPLICATE_PROVIDER')
     }
     const state: ProviderState = {
       provider,
-      active: true,
       chain: Promise.resolve(),
       liveIds: new Set(),
     }
     const dispose = ctx.effect(function* (this: SessionProviderCoordinator) {
       this._providers.set(provider.id, state)
-      yield () => {
-        state.active = false
+      yield async () => {
         this._providers.delete(provider.id)
+        await state.chain
       }
     }.bind(this), 'sessionQuery.registerSearchProvider()')
-    return () => void dispose()
+    return async () => { await dispose() }
   }
 
   /**
@@ -83,7 +81,7 @@ export class SessionProviderCoordinator {
   ): Promise<SessionSearchPage<SessionSearchHit>> {
     const state = this._resolveProvider()
     const normalized = this._normalizeSessionSearch(request)
-    const work = this._runFullSearch(state, async () => {
+    const work = this._runFullSearch(state, undefined, async () => {
       if (exec?.signal?.aborted) throw aborted()
       const result = await state.provider.searchSessions(normalized, exec)
       return this._validateSearchPage(state, result, normalized.limit)
@@ -113,22 +111,25 @@ export class SessionProviderCoordinator {
     if (live !== undefined) {
       work = this._runLiveSearch(state, live, query)
     } else {
-      const persistence = await this._corpus().persistenceView()
-      if (persistence === undefined || !persistence.headers.some(header => header.id === request.sessionId)) {
-        throw new SessionQueryError(`session "${request.sessionId}" not found`, 'SESSION_QUERY_SESSION_NOT_FOUND')
-      }
-      work = this._runFullSearch(state, query)
+      work = this._runFullSearch(state, request.sessionId, query)
     }
     return waitFor(work, exec?.signal)
   }
 
-  private _runFullSearch<T>(state: ProviderState, query: () => Promise<T>): Promise<T> {
+  private _runFullSearch<T>(
+    state: ProviderState,
+    requiredSessionId: SessionId | undefined,
+    query: () => Promise<T>,
+  ): Promise<T> {
     const liveSessions = this._corpus().listLive()
     return this._serialize(state, async () => {
       await this._synchronize(state, async () => {
-        /* v8 ignore next -- a provider can be disposed while queued behind an in-flight update */
-        if (!state.active) return
         const persistence = await this._corpus().persistenceView()
+        const missingRequired = requiredSessionId !== undefined
+          && (persistence === undefined || !persistence.headers.some(header => header.id === requiredSessionId))
+        if (missingRequired) {
+          throw new SessionQueryError(`session "${requiredSessionId}" not found`, 'SESSION_QUERY_SESSION_NOT_FOUND')
+        }
         if (persistence === undefined) {
           await state.provider.setPersistedActive(false)
         } else {
@@ -172,8 +173,6 @@ export class SessionProviderCoordinator {
     }
     return this._serialize(state, async () => {
       await this._synchronize(state, async () => {
-        /* v8 ignore next -- a provider can be disposed while queued behind an in-flight update */
-        if (!state.active) return
         await state.provider.replaceLive(snapshot)
         state.liveIds.add(session.id)
       })
@@ -272,32 +271,35 @@ export class SessionProviderCoordinator {
 }
 
 function waitFor<T>(work: Promise<T>, signal: AbortSignal | undefined): Promise<T> {
-  if (signal === undefined) return work
+  const observed = work.catch((error: unknown) => { throw operationError(error) })
+  if (signal === undefined) return observed
   if (signal.aborted) {
     // Cancellation supersedes the caller's result, but shared work must still
     // have a rejection observer when it has already failed synchronously.
-    void work.catch((_supersededError: unknown) => undefined)
+    void observed.catch((_supersededError: unknown) => undefined)
     return Promise.reject(aborted())
   }
   return new Promise<T>((resolve, reject) => {
     const onAbort = () => { reject(aborted()) }
     signal.addEventListener('abort', onAbort, { once: true })
-    work.then(
+    observed.then(
       (value) => {
         signal.removeEventListener('abort', onAbort)
         resolve(value)
       },
       (error: unknown) => {
         signal.removeEventListener('abort', onAbort)
-        /* v8 ignore next -- Promise contracts reject with Error; retain a typed boundary for third-party providers */
-        reject(error instanceof Error
-          ? error
-          : new SessionQueryError('session-query operation failed with a non-Error rejection', 'SESSION_QUERY_PROVIDER_ERROR', { cause: error }))
+        reject(operationError(error))
       },
     )
   })
 }
 
+function operationError(error: unknown): Error {
+  if (error instanceof Error) return error
+  return new SessionQueryError('session-query operation failed with a non-Error rejection', 'SESSION_QUERY_PROVIDER_ERROR', { cause: error })
+}
+
 function aborted(): SessionQueryError {
   return new SessionQueryError('session-query operation aborted', 'SESSION_QUERY_ABORTED')
 }

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

@@ -400,10 +400,37 @@ describe('provider selection and synchronization', () => {
     await expect(ctx.sessionQuery.searchEvents({ sessionId: session.id, query: 'x', limit: 1 }, { signal: new AbortController().signal }))
       .resolves.toMatchObject({ providerId: provider.id })
 
-    dispose()
+    await dispose()
     await expect(ctx.sessionQuery.searchSessions({ query: 'x' })).rejects.toThrow(expectCode('SESSION_QUERY_PROVIDER_UNAVAILABLE'))
   })
 
+  it('deselects immediately and drains accepted work before disposal settles', async () => {
+    const ctx = await liveContext()
+    const provider = new FakeProvider()
+    const queryStarted = deferred()
+    const releaseQuery = deferred()
+    provider.searchSessions = async () => {
+      queryStarted.resolve()
+      await releaseQuery.promise
+      return { providerId: provider.id, items: [] }
+    }
+    const dispose = ctx.sessionQuery.registerSearchProvider(provider)
+
+    const accepted = ctx.sessionQuery.searchSessions({ query: 'accepted' })
+    await queryStarted.promise
+    let disposed = false
+    const disposal = dispose().then(() => { disposed = true })
+    await expect(ctx.sessionQuery.searchSessions({ query: 'future' }))
+      .rejects.toThrow(expectCode('SESSION_QUERY_PROVIDER_UNAVAILABLE'))
+    await Promise.resolve()
+    expect(disposed).toBe(false)
+
+    releaseQuery.resolve()
+    await expect(accepted).resolves.toMatchObject({ providerId: provider.id })
+    await disposal
+    expect(disposed).toBe(true)
+  })
+
   it('serializes concurrent synchronization and supports cancellation while provider search is pending', async () => {
     const ctx = await liveContext()
     const session = ctx.sessions.create(SessionId('coalesce'))
@@ -559,6 +586,60 @@ describe('provider selection and synchronization', () => {
     expect(provider.persisted.get(persisted.id)?.documents[0]?.text).toBe('persisted target')
   })
 
+  it('cancels a persisted-only event search while persistence listing is blocked', async () => {
+    const persisted = header('blocked-persisted-target', 1)
+    TestPersistence.reset([{ meta: persisted, events: eventLog('persisted target') }])
+    const listStarted = deferred()
+    const releaseList = deferred()
+    TestPersistence.onList = listStarted.resolve
+    TestPersistence.listBarrier = releaseList.promise
+    const ctx = await liveContext()
+    const persistenceFiber = await ctx.plugin(TestPersistence)
+    await listStarted.promise
+    const provider = new FakeProvider()
+    const disposeProvider = ctx.sessionQuery.registerSearchProvider(provider)
+    const controller = new AbortController()
+
+    const pending = ctx.sessionQuery.searchEvents(
+      { sessionId: persisted.id, query: 'target' },
+      { signal: controller.signal },
+    )
+    controller.abort()
+    await expect(pending).rejects.toThrow(expectCode('SESSION_QUERY_ABORTED'))
+    expect(provider.eventRequests).toEqual([])
+
+    releaseList.resolve()
+    await disposeProvider()
+    await persistenceFiber.dispose()
+    TestPersistence.listBarrier = undefined
+    TestPersistence.onList = undefined
+  })
+
+  it('normalizes non-Error query rejections and preserves Error identity', async () => {
+    const ctx = await liveContext()
+    const provider = new FakeProvider()
+    ctx.sessionQuery.registerSearchProvider(provider)
+    const signals = [undefined, new AbortController().signal]
+
+    for (const [index, signal] of signals.entries()) {
+      const exec = signal === undefined ? undefined : { signal }
+      const identity = new Error(`query failure ${index}`)
+      provider.searchSessions = () => Promise.reject(identity)
+      const preserved = await ctx.sessionQuery.searchSessions({ query: 'x' }, exec)
+        .then(() => undefined, (error: unknown) => error)
+      expect(preserved).toBe(identity)
+
+      const rejection = { index }
+      // Deliberately violate the Promise convention to test the provider boundary.
+      // eslint-disable-next-line @typescript-eslint/prefer-promise-reject-errors
+      provider.searchSessions = () => Promise.reject(rejection)
+      const normalized = await ctx.sessionQuery.searchSessions({ query: 'x' }, exec)
+        .then(() => undefined, (error: unknown) => error)
+      expect(normalized).toBeInstanceOf(SessionQueryError)
+      expect(normalized).toMatchObject({ code: 'SESSION_QUERY_PROVIDER_ERROR', cause: rejection })
+    }
+  })
+
   it('fails loudly for duplicate, configured, unavailable, and ambiguous providers', async () => {
     const ctx = await liveContext()
     const first = new FakeProvider('first')