From 8dd85bd1852ca7fc2a3f481c561bdc7184d001a0 Mon Sep 17 00:00:00 2001 From: Hypatia May Date: Sat, 11 Jul 2026 11:48:32 +0800 Subject: [PATCH] refactor(session-query): simplify provider synchronization --- docs/event-producer-consumer.md | 6 +- .../session-query/session-query/src/corpus.ts | 9 +- .../session-query/src/extraction.ts | 6 +- .../session-query/session-query/src/index.ts | 6 +- .../session-query/src/provider.ts | 154 ++++++------------ .../session-query/tests/session-query.spec.ts | 63 +++++-- 6 files changed, 106 insertions(+), 138 deletions(-) diff --git a/docs/event-producer-consumer.md b/docs/event-producer-consumer.md index 0b8c4fce02..23e3844375 100644 --- a/docs/event-producer-consumer.md +++ b/docs/event-producer-consumer.md @@ -23,11 +23,11 @@ This matrix shows which packages dispatch each harness-owned event and which pac | `fs/observed` | `emit` | [`packages/fs/fs/src/index.ts:138`](../packages/fs/fs/src/index.ts) | [`tool-fs`](../packages/fs/tool-fs) (`emit`) | [`fs-policy`](../packages/fs/fs-policy) | | `fs/write-intent` | `waterfall` | [`packages/fs/fs/src/index.ts:109`](../packages/fs/fs/src/index.ts) | [`tool-fs`](../packages/fs/tool-fs) (`waterfall`) | [`fs-policy`](../packages/fs/fs-policy) | | `llm/stream` | `waterfall` | [`packages/llm/llm/src/index.ts:39`](../packages/llm/llm/src/index.ts) | [`llm`](../packages/llm/llm) (`waterfall`) | [`invariants`](../packages/support/invariants), [`llm-replay`](../packages/support/llm-replay) | -| `session/created` | `emit` | [`packages/core/session/src/index.ts:39`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`emit`) | [`invariants`](../packages/support/invariants), [`session-persistence`](../packages/session-persistence/session-persistence), [`session-query`](../packages/session-query/session-query) | -| `session/event` | `emit` | [`packages/core/session/src/index.ts:55`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`emit`) | [`acp`](../packages/ui/acp), [`invariants`](../packages/support/invariants), [`session-persistence`](../packages/session-persistence/session-persistence), [`session-query`](../packages/session-query/session-query), [`stdio-agent`](../packages/ui/stdio-agent) | +| `session/created` | `emit` | [`packages/core/session/src/index.ts:39`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`emit`) | [`invariants`](../packages/support/invariants), [`session-persistence`](../packages/session-persistence/session-persistence) | +| `session/event` | `emit` | [`packages/core/session/src/index.ts:55`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`emit`) | [`acp`](../packages/ui/acp), [`invariants`](../packages/support/invariants), [`session-persistence`](../packages/session-persistence/session-persistence), [`stdio-agent`](../packages/ui/stdio-agent) | | `session/flush` | `parallel` | [`packages/core/session/src/index.ts:65`](../packages/core/session/src/index.ts) | [`agent-loop`](../packages/core/agent-loop) (`parallel`) | [`session-persistence`](../packages/session-persistence/session-persistence) | | `session/persisted` | `parallel` | [`packages/session-persistence/session-persistence/src/index.ts:50`](../packages/session-persistence/session-persistence/src/index.ts) | [`session-persistence`](../packages/session-persistence/session-persistence) (`parallel`) | [`session-query`](../packages/session-query/session-query) | -| `session/removed` | `parallel` | [`packages/core/session/src/index.ts:47`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`parallel`) | [`session-query`](../packages/session-query/session-query) | +| `session/removed` | `parallel` | [`packages/core/session/src/index.ts:47`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`parallel`) | - | | `subagent/end` | `emit` | [`packages/subagent/subagent/src/index.ts:98`](../packages/subagent/subagent/src/index.ts) | [`subagent`](../packages/subagent/subagent) (`events.dispatch`) | [`hooks-claude`](../packages/hooks/hooks-claude) | | `subagent/provider-added` | `emit` | [`packages/subagent/subagent/src/index.ts:72`](../packages/subagent/subagent/src/index.ts) | [`subagent`](../packages/subagent/subagent) (`emit`) | [`tool-subagent`](../packages/subagent/tool-subagent) | | `subagent/provider-removed` | `emit` | [`packages/subagent/subagent/src/index.ts:83`](../packages/subagent/subagent/src/index.ts) | - | [`tool-subagent`](../packages/subagent/tool-subagent) | diff --git a/packages/session-query/session-query/src/corpus.ts b/packages/session-query/session-query/src/corpus.ts index 87fb7a5533..84d6967e01 100644 --- a/packages/session-query/session-query/src/corpus.ts +++ b/packages/session-query/session-query/src/corpus.ts @@ -36,10 +36,7 @@ export interface PersistenceView { export class SessionCorpus { private _persistence: PersistenceBinding | undefined - constructor( - private readonly _ctx: Context, - private readonly _onPersistenceChange: (active: boolean) => void, - ) { + constructor(private readonly _ctx: Context) { _ctx.effect(() => { const fiber = _ctx.inject(['sessionPersistence'], (childCtx: Context) => { this._attachPersistence(childCtx, childCtx.sessionPersistence) @@ -145,7 +142,6 @@ export class SessionCorpus { refreshing: undefined, } this._persistence = binding - this._onPersistenceChange(true) void this._refreshPersistence(binding) ctx.on('session/persisted', (header) => { /* v8 ignore next -- a stale notification can race optional-service disposal */ @@ -154,7 +150,6 @@ export class SessionCorpus { const observation = { generation: ++binding.observationGeneration, header: snapshot } binding.headers.set(header.id, snapshot) binding.observations.set(header.id, observation) - this._onPersistenceChange(true) }) ctx.effect(() => () => { this._detachPersistence(binding) }, 'sessionQuery.persistenceBinding') } @@ -163,7 +158,6 @@ export class SessionCorpus { /* v8 ignore next -- duplicate optional-service disposal is a Cordis teardown edge */ if (this._persistence?.token !== binding.token) return this._persistence = undefined - this._onPersistenceChange(false) } private _refreshPersistence(binding: PersistenceBinding): Promise { @@ -184,7 +178,6 @@ export class SessionCorpus { } binding.headers = nextHeaders binding.error = undefined - this._onPersistenceChange(true) }).catch((error: unknown) => { /* v8 ignore next -- a failed list can race optional-service disposal */ if (this._persistence?.token !== binding.token) return diff --git a/packages/session-query/session-query/src/extraction.ts b/packages/session-query/session-query/src/extraction.ts index 58db38c8ef..c4ad294a4f 100644 --- a/packages/session-query/session-query/src/extraction.ts +++ b/packages/session-query/session-query/src/extraction.ts @@ -37,7 +37,7 @@ export class SessionTextExtractors { private readonly _eventExtractors = new Map() private readonly _contentExtractors = new Map() - constructor(private readonly _onChange: () => void) { + constructor() { this._installCoreExtractors() } @@ -63,10 +63,8 @@ export class SessionTextExtractors { } const dispose = ctx.effect(function* (this: SessionTextExtractors) { this._eventExtractors.set(type, stored) - this._onChange() yield () => { this._eventExtractors.delete(type) - this._onChange() } }.bind(this), `sessionQuery.eventExtractor(${type})`) return () => void dispose() @@ -94,10 +92,8 @@ export class SessionTextExtractors { } const dispose = ctx.effect(function* (this: SessionTextExtractors) { this._contentExtractors.set(type, stored) - this._onChange() yield () => { this._contentExtractors.delete(type) - this._onChange() } }.bind(this), `sessionQuery.contentExtractor(${type})`) return () => void dispose() diff --git a/packages/session-query/session-query/src/index.ts b/packages/session-query/session-query/src/index.ts index 1145332c23..edab597dd5 100644 --- a/packages/session-query/session-query/src/index.ts +++ b/packages/session-query/session-query/src/index.ts @@ -78,13 +78,13 @@ export class SessionQueryService extends Service { if (defaultLimit > maxLimit) { throw new SessionQueryError('session-query: defaultLimit must be <= maxLimit', 'SESSION_QUERY_INVALID_CONFIG') } - this._extractors = new SessionTextExtractors(() => { this._providers.invalidateAll() }) - this._providers = new SessionProviderCoordinator(ctx, { + this._extractors = new SessionTextExtractors() + this._providers = new SessionProviderCoordinator({ ...config.searchProvider !== undefined ? { searchProvider: config.searchProvider } : {}, defaultLimit, maxLimit, }, () => this._corpus, this._extractors) - this._corpus = new SessionCorpus(ctx, (active) => { this._providers.persistenceChanged(active) }) + this._corpus = new SessionCorpus(ctx) } /** diff --git a/packages/session-query/session-query/src/provider.ts b/packages/session-query/session-query/src/provider.ts index 7152e9f5e5..c3ec4ad0f4 100644 --- a/packages/session-query/session-query/src/provider.ts +++ b/packages/session-query/session-query/src/provider.ts @@ -26,13 +26,6 @@ interface ProviderState { active: boolean chain: Promise liveIds: Set - fullSync: FullSync | undefined - liveSync: Map> -} - -interface FullSync { - liveKey: string - promise: Promise } /** Coordinates one selected provider against live and persisted corpus layers. */ @@ -43,7 +36,6 @@ export class SessionProviderCoordinator { private readonly _providers = new Map() constructor( - private readonly _ctx: Context, config: Required> & Pick, private readonly _corpus: () => SessionCorpus, private readonly _extractors: SessionTextExtractors, @@ -51,9 +43,6 @@ export class SessionProviderCoordinator { this._configuredProviderId = config.searchProvider this._defaultLimit = config.defaultLimit this._maxLimit = config.maxLimit - _ctx.on('session/created', (session) => { this.invalidateLive(session.id) }) - _ctx.on('session/event', (session) => { this.invalidateLive(session.id) }) - _ctx.on('session/removed', (header) => { this.invalidateLive(header.id) }) } /** @@ -71,14 +60,9 @@ export class SessionProviderCoordinator { active: true, chain: Promise.resolve(), liveIds: new Set(), - fullSync: undefined, - liveSync: new Map(), } const dispose = ctx.effect(function* (this: SessionProviderCoordinator) { this._providers.set(provider.id, state) - void this._enqueue(state, () => provider.setPersistedActive(false)).catch((error: unknown) => { - this._ctx.logger.warn(`session-query provider "${provider.id}" failed initial deactivation: ${String(error)}`) - }) yield () => { state.active = false this._providers.delete(provider.id) @@ -99,9 +83,12 @@ export class SessionProviderCoordinator { ): Promise> { const state = this._resolveProvider() const normalized = this._normalizeSessionSearch(request) - await waitFor(this._syncAll(state), exec?.signal) - const result = await waitFor(state.provider.searchSessions(normalized, exec), exec?.signal) - return this._validateSearchPage(state, result, normalized.limit) + const work = this._runFullSearch(state, async () => { + if (exec?.signal?.aborted) throw aborted() + const result = await state.provider.searchSessions(normalized, exec) + return this._validateSearchPage(state, result, normalized.limit) + }) + return waitFor(work, exec?.signal) } /** @@ -116,87 +103,41 @@ export class SessionProviderCoordinator { ): Promise> { const state = this._resolveProvider() const normalized = this._normalizeEventSearch(request) + const query = async (): Promise> => { + if (exec?.signal?.aborted) throw aborted() + const result = await state.provider.searchEvents(normalized, exec) + return this._validateSearchPage(state, result, normalized.limit) + } const live = this._corpus().getLive(request.sessionId) + let work: Promise> if (live !== undefined) { - await waitFor(this._syncLive(state, live), exec?.signal) + 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') } - await waitFor(this._syncAll(state), exec?.signal) + work = this._runFullSearch(state, query) } - const result = await waitFor(state.provider.searchEvents(normalized, exec), exec?.signal) - return this._validateSearchPage(state, result, normalized.limit) + return waitFor(work, exec?.signal) } - /** - * Invalidate provider synchronization after one live source change. - * @param sessionId - changed live session. - */ - invalidateLive(sessionId: SessionId): void { - for (const state of this._providers.values()) { - state.fullSync = undefined - state.liveSync.delete(sessionId) - } - } - - /** Invalidate all source/extractor-derived provider snapshots. */ - invalidateAll(): void { - for (const state of this._providers.values()) { - state.fullSync = undefined - state.liveSync.clear() - } - } - - /** - * React to persistence mount, inventory change, or unmount. - * @param active - whether canonical persistence remains mounted. - */ - persistenceChanged(active: boolean): void { - for (const state of this._providers.values()) state.fullSync = undefined - if (active) return - for (const state of this._providers.values()) { - void this._enqueue(state, () => state.provider.setPersistedActive(false)).catch((error: unknown) => { - this._ctx.logger.warn(`session-query provider "${state.provider.id}" failed persistence deactivation: ${String(error)}`) + private _runFullSearch(state: ProviderState, query: () => Promise): Promise { + 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() + if (persistence === undefined) { + await state.provider.setPersistedActive(false) + } else { + await this._syncPersisted(state, persistence) + } + await this._replaceLiveCorpus(state, liveSessions) }) - } - } - - private _syncAll(state: ProviderState): Promise { - // Capture the direct source before awaiting: only searches that observed - // the same live corpus may share an in-flight full synchronization. - let liveSessions: Session[] - let liveKey: string - try { - liveSessions = this._corpus().listLive() - liveKey = JSON.stringify(liveSessions.map(session => this._snapshotLive(session)).map(snapshot => [ - snapshot.session.header.id, - snapshot.fingerprint, - snapshot.session.persisted, - ])) - } catch (error: unknown) { - return Promise.reject(this._synchronizationError(state, error)) - } - 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 - const persistence = await this._corpus().persistenceView() - if (persistence === undefined) { - await state.provider.setPersistedActive(false) - } else { - await this._syncPersisted(state, persistence) - } - await this._replaceLiveCorpus(state, liveSessions) + return query() }) - 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 === fullSync) state.fullSync = undefined - }).catch(() => undefined) - return promise } private async _syncPersisted(state: ProviderState, persistence: PersistenceView): Promise { @@ -222,39 +163,42 @@ export class SessionProviderCoordinator { state.liveIds = liveIds } - private _syncLive(state: ProviderState, session: Session): Promise { - const existing = state.liveSync.get(session.id) - if (existing !== undefined) return existing + private _runLiveSearch(state: ProviderState, session: Session, query: () => Promise): Promise { let snapshot: ReturnType try { snapshot = this._snapshotLive(session) } catch (error: unknown) { return Promise.reject(this._synchronizationError(state, error)) } - 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 - await state.provider.replaceLive(snapshot) - state.liveIds.add(session.id) + 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) + }) + return query() }) - state.liveSync.set(session.id, promise) - void promise.finally(() => { - /* v8 ignore next -- a newer invalidation may already own the target slot */ - if (state.liveSync.get(session.id) === promise) state.liveSync.delete(session.id) - }).catch(() => undefined) - return promise } private _snapshotLive(session: Session): ReturnType { return this._extractors.buildSnapshot(this._corpus().snapshotLive(session)) } - private _enqueue(state: ProviderState, operation: () => Promise): Promise { + /** Serialize reconciliation and its provider query as one stable transaction. */ + private _serialize(state: ProviderState, operation: () => Promise): Promise { const next = state.chain.then(operation, operation) state.chain = next.then(() => undefined, () => undefined) - return next.catch((error: unknown) => { + return next + } + + /** Translate only derived-index update failures, never provider query failures. */ + private async _synchronize(state: ProviderState, operation: () => Promise): Promise { + try { + await operation() + } catch (error: unknown) { throw this._synchronizationError(state, error) - }) + } } private _synchronizationError(state: ProviderState, error: unknown): SessionQueryError { diff --git a/packages/session-query/session-query/tests/session-query.spec.ts b/packages/session-query/session-query/tests/session-query.spec.ts index 5249cd22ae..642bfb263f 100644 --- a/packages/session-query/session-query/tests/session-query.spec.ts +++ b/packages/session-query/session-query/tests/session-query.spec.ts @@ -103,7 +103,6 @@ class FakeProvider implements SessionSearchProvider { eventRequests: SessionEventSearchSpec[] = [] failNextLive = false failNextPersisted = false - failNextActive = false sessionPage: SessionSearchPage eventPage: SessionSearchPage @@ -125,10 +124,6 @@ class FakeProvider implements SessionSearchProvider { } setPersistedActive(active: boolean): Promise { - if (this.failNextActive) { - this.failNextActive = false - return Promise.reject(new Error('activation failed')) - } this.activeHistory.push(active) return Promise.resolve() } @@ -409,7 +404,7 @@ describe('provider selection and synchronization', () => { await expect(ctx.sessionQuery.searchSessions({ query: 'x' })).rejects.toThrow(expectCode('SESSION_QUERY_PROVIDER_UNAVAILABLE')) }) - it('coalesces concurrent synchronization and supports cancellation while provider search is pending', async () => { + it('serializes concurrent synchronization and supports cancellation while provider search is pending', async () => { const ctx = await liveContext() const session = ctx.sessions.create(SessionId('coalesce')) session.append('user/message', { content: [{ type: 'text', text: 'x' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) @@ -418,34 +413,40 @@ describe('provider selection and synchronization', () => { let releaseLive!: () => void const liveBarrier = new Promise((resolve) => { releaseLive = resolve }) + const liveStarted = deferred() let replacements = 0 provider.replaceLive = async (snapshot) => { replacements += 1 + liveStarted.resolve() await liveBarrier provider.live.set(snapshot.session.header.id, structuredClone(snapshot)) } const first = ctx.sessionQuery.searchEvents({ sessionId: session.id, query: 'x' }) const second = ctx.sessionQuery.searchEvents({ sessionId: session.id, query: 'x' }) - await Promise.resolve() + await liveStarted.promise + expect(replacements).toBe(1) releaseLive() await Promise.all([first, second]) - expect(replacements).toBe(1) + expect(replacements).toBe(2) session.append('user/message', { content: [{ type: 'text', text: 'y' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) let releaseCorpus!: () => void const corpusBarrier = new Promise((resolve) => { releaseCorpus = resolve }) + const corpusStarted = deferred() let corpusReplacements = 0 provider.replaceLive = async (snapshot) => { corpusReplacements += 1 + corpusStarted.resolve() await corpusBarrier provider.live.set(snapshot.session.header.id, structuredClone(snapshot)) } const crossFirst = ctx.sessionQuery.searchSessions({ query: 'x' }) const crossSecond = ctx.sessionQuery.searchSessions({ query: 'x' }) - await Promise.resolve() + await corpusStarted.promise + expect(corpusReplacements).toBe(1) releaseCorpus() await Promise.all([crossFirst, crossSecond]) - expect(corpusReplacements).toBe(1) + expect(corpusReplacements).toBe(2) let releaseSearch!: () => void const searchBarrier = new Promise((resolve) => { releaseSearch = resolve }) @@ -467,6 +468,42 @@ describe('provider selection and synchronization', () => { .rejects.toThrow('search failed') }) + it('holds a provider query stable until later reconciliation can begin', async () => { + const ctx = await liveContext() + const session = ctx.sessions.create(SessionId('stable-query')) + session.append('user/message', { content: [{ type: 'text', text: 'stable' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) + const provider = new FakeProvider() + const queryStarted = deferred() + const releaseQuery = deferred() + provider.searchEvents = async () => { + queryStarted.resolve() + await releaseQuery.promise + return { providerId: provider.id, items: [] } + } + const reconciliationStarted = deferred() + let reconciling = false + provider.setPersistedActive = (active) => { + if (!active) { + reconciling = true + reconciliationStarted.resolve() + } + return Promise.resolve() + } + ctx.sessionQuery.registerSearchProvider(provider) + + const eventSearch = ctx.sessionQuery.searchEvents({ sessionId: session.id, query: 'stable' }) + await queryStarted.promise + const fullSearch = ctx.sessionQuery.searchSessions({ query: 'stable' }) + await new Promise((resolve) => { setImmediate(resolve) }) + expect(reconciling).toBe(false) + + releaseQuery.resolve() + await eventSearch + await reconciliationStarted.promise + expect(reconciling).toBe(true) + await fullSearch + }) + 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')) @@ -563,7 +600,6 @@ describe('provider selection and synchronization', () => { live.append('user/message', { content: [{ type: 'text', text: 'override' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) const persistenceFiber = await ctx.plugin(TestPersistence) const provider = new FakeProvider() - provider.failNextActive = true provider.persisted.set(SessionId('stale'), { session: { header: header('stale'), live: false, persisted: true }, fingerprint: 'stale', documents: [] }) ctx.sessionQuery.registerSearchProvider(provider) @@ -583,7 +619,6 @@ describe('provider selection and synchronization', () => { await ctx.sessionQuery.searchSessions({ query: 'x' }) expect(provider.persisted.get(announced.id)?.documents[0]?.text).toBe('announced') - provider.failNextActive = true await persistenceFiber.dispose() await ctx.sessionQuery.searchSessions({ query: 'x' }) expect(provider.activeHistory.at(-1)).toBe(false) @@ -674,7 +709,7 @@ describe('provider selection and synchronization', () => { expect(provider.persisted.get(persisted.id)?.documents[0]?.text).toBe('retry') }) - it('types synchronous extractor failures during full-search key construction', async () => { + it('types extractor failures during queued full synchronization', async () => { const ctx = await liveContext() const session = ctx.sessions.create(SessionId('throwing-extractor')) session.append('test/note', { note: 'unreachable' }) @@ -714,7 +749,7 @@ describe('provider selection and synchronization', () => { const onUnhandled = (reason: unknown) => { unhandled.push(reason) } process.on('unhandledRejection', onUnhandled) try { - await expect(ctx.sessionQuery.searchSessions({ query: 'x' }, { signal: controller.signal })) + await expect(ctx.sessionQuery.searchEvents({ sessionId: session.id, query: 'x' }, { signal: controller.signal })) .rejects.toThrow(expectCode('SESSION_QUERY_ABORTED')) await new Promise((resolve) => { setImmediate(resolve) }) expect(unhandled).toEqual([])