From dc1c236cc04cb15cac356f4e19326d919f169812 Mon Sep 17 00:00:00 2001 From: digua Date: Mon, 22 Jun 2026 23:59:01 +0800 Subject: [PATCH] fix: address semantic worker and pull review bugs --- apps/desktop/main/internal-api.ts | 4 +- .../src/semantic-index/worker-client.test.ts | 197 +++++++++++++++++- .../src/semantic-index/worker-client.ts | 144 ++++++++++--- packages/sync/src/pull-engine.test.ts | 103 +++++++++ packages/sync/src/pull-engine.ts | 9 +- 5 files changed, 423 insertions(+), 34 deletions(-) diff --git a/apps/desktop/main/internal-api.ts b/apps/desktop/main/internal-api.ts index c1bc2016..ca772a2c 100644 --- a/apps/desktop/main/internal-api.ts +++ b/apps/desktop/main/internal-api.ts @@ -268,7 +268,7 @@ export async function startInternalServer(pathProvider: PathProvider): Promise { /* best-effort */ } try { - semanticIndexService?.close() + await semanticIndexService?.close() } catch { /* best-effort */ } diff --git a/packages/node-runtime/src/semantic-index/worker-client.test.ts b/packages/node-runtime/src/semantic-index/worker-client.test.ts index 0b222679..224e70aa 100644 --- a/packages/node-runtime/src/semantic-index/worker-client.test.ts +++ b/packages/node-runtime/src/semantic-index/worker-client.test.ts @@ -3,7 +3,8 @@ import fs from 'node:fs' import os from 'node:os' import path from 'node:path' import test from 'node:test' -import { SemanticIndexConfigStore } from './config' +import { SemanticIndexConfigStore, type SemanticIndexConfig } from './config' +import type { SemanticIndexSessionStatus } from './service' import { createSemanticIndexWorkerClient, type SemanticIndexWorkerTransport, @@ -21,11 +22,17 @@ class FakeTransport implements SemanticIndexWorkerTransport { requests: Array<{ method: string; args: unknown[] }> = [] closed = false + constructor(private readonly handler?: (method: string, args: unknown[]) => unknown) {} + async request(method: string, args: unknown[]): Promise { this.requests.push({ method, args }) + if (this.handler) return (await this.handler(method, args)) as T if (method === 'status') { return null as T } + if (method === 'setConfig') { + return args[0] as T + } throw new Error(`unexpected method: ${method}`) } @@ -34,14 +41,35 @@ class FakeTransport implements SemanticIndexWorkerTransport { } } -function makeFactory(instances: FakeTransport[]): SemanticIndexWorkerTransportFactory { +function makeFactory( + instances: FakeTransport[], + handler?: (method: string, args: unknown[]) => unknown +): SemanticIndexWorkerTransportFactory { return () => { - const transport = new FakeTransport() + const transport = new FakeTransport(handler) instances.push(transport) return transport } } +function makeStatus(sessionId: string, active: boolean): SemanticIndexSessionStatus { + return { + sessionId, + enabled: true, + indexStatus: active ? 'running' : 'completed', + needsRebuild: false, + totalMessages: 10, + indexedMessages: active ? 5 : 10, + chunkCount: active ? 1 : 2, + coverage: active ? 0.5 : 1, + queued: false, + running: active, + partial: active, + error: null, + modelId: 'model-a', + } +} + test('worker client keeps config-only operations in process without starting worker', async () => { const instances: FakeTransport[] = [] const client = createSemanticIndexWorkerClient({ @@ -58,6 +86,56 @@ test('worker client keeps config-only operations in process without starting wor assert.equal(instances.length, 0) }) +test('worker client returns unavailable for unconfigured canSearch without starting worker', async () => { + const instances: FakeTransport[] = [] + const client = createSemanticIndexWorkerClient({ + configStore: makeConfigStore(), + transportFactory: makeFactory(instances), + idleTimeoutMs: 1000, + }) + + const available = await client.canSearch('session-a') + + assert.equal(available, false) + assert.equal(instances.length, 0) +}) + +test('worker client treats canSearch worker failures as unavailable', async () => { + const instances: FakeTransport[] = [] + const configStore = makeConfigStore() + configStore.set({ version: 1, mode: 'local', local: { modelId: 'model-a' }, api: null }) + const client = createSemanticIndexWorkerClient({ + configStore, + transportFactory: makeFactory(instances, () => { + throw new Error('worker failed to start') + }), + idleTimeoutMs: 1000, + }) + + const available = await client.canSearch('session-a') + + assert.equal(available, false) + assert.equal(instances.length, 1) + assert.equal(instances[0].closed, true) +}) + +test('worker client treats status worker failures as unavailable', async () => { + const instances: FakeTransport[] = [] + const client = createSemanticIndexWorkerClient({ + configStore: makeConfigStore(), + transportFactory: makeFactory(instances, () => { + throw new Error('worker failed to start') + }), + idleTimeoutMs: 1000, + }) + + const status = await client.status('session-a') + + assert.equal(status, null) + assert.equal(instances.length, 1) + assert.equal(instances[0].closed, true) +}) + test('worker client lazy-starts for runtime calls and closes after idle timeout', async () => { const instances: FakeTransport[] = [] const timers: Array<() => void> = [] @@ -89,3 +167,116 @@ test('worker client lazy-starts for runtime calls and closes after idle timeout' assert.equal(instances[0].closed, true) }) + +test('worker client forwards config updates to a live worker without starting a new one', async () => { + const instances: FakeTransport[] = [] + const client = createSemanticIndexWorkerClient({ + configStore: makeConfigStore(), + transportFactory: makeFactory(instances), + idleTimeoutMs: 1000, + }) + + await client.status('session-a') + const initial = await client.getConfig() + const updated: SemanticIndexConfig = { ...initial, enabled: false } + + const saved = await client.setConfig(updated) + + assert.equal(saved.enabled, false) + assert.equal(instances.length, 1) + assert.deepEqual(instances[0].requests, [ + { method: 'status', args: ['session-a'] }, + { method: 'setConfig', args: [saved] }, + ]) +}) + +test('worker client clears tracked active builds when semantic indexing is disabled', async () => { + const instances: FakeTransport[] = [] + const timers: Array<() => void> = [] + const client = createSemanticIndexWorkerClient({ + configStore: makeConfigStore(), + transportFactory: makeFactory(instances, (method, args) => { + if (method === 'build') return undefined + if (method === 'buildAllPending') return undefined + if (method === 'setConfig') return args[0] + throw new Error(`unexpected method: ${method}`) + }), + idleTimeoutMs: 1000, + timers: { + setTimeout(callback) { + timers.push(callback) + return callback + }, + clearTimeout() { + /* test timer is manually triggered */ + }, + }, + }) + + await client.build('session-a') + await client.buildAllPending() + const initial = await client.getConfig() + await client.setConfig({ ...initial, enabled: false }) + + assert.equal(timers.length, 1) + timers.shift()?.() + assert.equal(instances[0].closed, true) +}) + +test('worker client does not clear active-build state from partial status snapshots', async () => { + const instances: FakeTransport[] = [] + const timers: Array<() => void> = [] + const client = createSemanticIndexWorkerClient({ + configStore: makeConfigStore(), + transportFactory: makeFactory(instances, (method) => { + if (method === 'listEnabledStatuses') return [makeStatus('session-a', true)] + if (method === 'status') return makeStatus('session-b', false) + throw new Error(`unexpected method: ${method}`) + }), + idleTimeoutMs: 1000, + timers: { + setTimeout(callback) { + timers.push(callback) + return callback + }, + clearTimeout() { + /* test timer is manually triggered */ + }, + }, + }) + + await client.listEnabledStatuses() + await client.status('session-b') + + assert.equal(timers.length, 0) +}) + +test('worker client clears session active-build state from matching status snapshots', async () => { + const instances: FakeTransport[] = [] + const timers: Array<() => void> = [] + const client = createSemanticIndexWorkerClient({ + configStore: makeConfigStore(), + transportFactory: makeFactory(instances, (method, args) => { + if (method === 'build') return undefined + if (method === 'status') return makeStatus(args[0] as string, false) + throw new Error(`unexpected method: ${method}`) + }), + idleTimeoutMs: 1000, + timers: { + setTimeout(callback) { + timers.push(callback) + return callback + }, + clearTimeout() { + /* test timer is manually triggered */ + }, + }, + }) + + await client.build('session-a') + await client.status('session-a') + + assert.equal(timers.length, 1) + timers.shift()?.() + assert.equal(instances[0].closed, true) +}) diff --git a/packages/node-runtime/src/semantic-index/worker-client.ts b/packages/node-runtime/src/semantic-index/worker-client.ts index 91ffd857..5c34c96a 100644 --- a/packages/node-runtime/src/semantic-index/worker-client.ts +++ b/packages/node-runtime/src/semantic-index/worker-client.ts @@ -3,7 +3,7 @@ import type { PathProvider } from '@openchatlab/core' import type { AuthProfile } from '@openchatlab/config' import { resolveApiKey as defaultResolveApiKey } from '@openchatlab/config' import type { SemanticIndexConfig, SemanticIndexConfigInput } from './config' -import { SemanticIndexConfigStore } from './config' +import { isKeylessSemanticIndexApiBaseUrl, SemanticIndexConfigStore } from './config' import { SEMANTIC_INDEX_CONFIG_FILE, persistSemanticIndexConfig, @@ -61,7 +61,8 @@ export class SemanticIndexWorkerClient implements SemanticIndexRuntime { private transport: SemanticIndexWorkerTransport | null = null private pendingRequests = 0 private idleTimer: unknown = null - private hasActiveBuild = false + private activeBuildSessionIds = new Set() + private hasUnknownActiveBuild = false private readonly configStore: SemanticIndexConfigStore private readonly transportFactory: SemanticIndexWorkerTransportFactory private readonly idleTimeoutMs: number @@ -82,11 +83,14 @@ export class SemanticIndexWorkerClient implements SemanticIndexRuntime { return this.configStore.get() } - setConfig(config: SemanticIndexConfigInput, options?: { apiKey?: string }): SemanticIndexConfig { - return persistSemanticIndexConfig(this.configStore, config, { + async setConfig(config: SemanticIndexConfigInput, options?: { apiKey?: string }): Promise { + const saved = persistSemanticIndexConfig(this.configStore, config, { apiKey: options?.apiKey, writeAuthProfile: this.writeAuthProfile, }) + if (!saved.enabled) this.clearActiveBuildTracking() + if (!this.transport) return saved + return await this.call('setConfig', [saved]) } isConfigured(): boolean { @@ -105,9 +109,14 @@ export class SemanticIndexWorkerClient implements SemanticIndexRuntime { return this.call('disable', [sessionId]) } - build(sessionId: string): Promise { - this.hasActiveBuild = true - return this.call('build', [sessionId]) + async build(sessionId: string): Promise { + this.activeBuildSessionIds.add(sessionId) + try { + await this.call('build', [sessionId]) + } catch (error) { + this.activeBuildSessionIds.delete(sessionId) + throw error + } } pause(sessionId: string): Promise { @@ -118,36 +127,56 @@ export class SemanticIndexWorkerClient implements SemanticIndexRuntime { return this.call('cancel', [sessionId]) } - rebuild(sessionId: string): Promise { - this.hasActiveBuild = true - return this.call('rebuild', [sessionId]) + async rebuild(sessionId: string): Promise { + this.activeBuildSessionIds.add(sessionId) + try { + await this.call('rebuild', [sessionId]) + } catch (error) { + this.activeBuildSessionIds.delete(sessionId) + throw error + } } - buildAllPending(): Promise { - this.hasActiveBuild = true - return this.call('buildAllPending', []) + async buildAllPending(): Promise { + this.hasUnknownActiveBuild = true + try { + await this.call('buildAllPending', []) + } catch (error) { + this.hasUnknownActiveBuild = false + throw error + } } async listEnabledStatuses(): Promise { - const statuses = await this.call('listEnabledStatuses', []) - this.updateActiveBuildFromStatuses(statuses) + const statuses = await this.callProbe('listEnabledStatuses', [], [], { + scheduleIdle: false, + }) + this.updateActiveBuildFromCompleteSnapshot(statuses) + this.scheduleIdleCloseIfNeeded() return statuses } async status(sessionId: string): Promise { - const status = await this.call('status', [sessionId]) - this.updateActiveBuildFromStatuses(status ? [status] : []) + const status = await this.callProbe('status', [sessionId], null, { + scheduleIdle: false, + }) + this.updateActiveBuildFromPartialSnapshot([sessionId], status ? [status] : []) + this.scheduleIdleCloseIfNeeded() return status } async statusForSessions(sessionIds: string[]): Promise { - const statuses = await this.call('statusForSessions', [sessionIds]) - this.updateActiveBuildFromStatuses(statuses) + const statuses = await this.callProbe('statusForSessions', [sessionIds], [], { + scheduleIdle: false, + }) + this.updateActiveBuildFromPartialSnapshot(sessionIds, statuses) + this.scheduleIdleCloseIfNeeded() return statuses } - canSearch(sessionId: string): Promise { - return this.call('canSearch', [sessionId]) + async canSearch(sessionId: string): Promise { + if (!this.canRunFromLocalConfig()) return false + return await this.callProbe('canSearch', [sessionId], false) } search( @@ -179,7 +208,7 @@ export class SemanticIndexWorkerClient implements SemanticIndexRuntime { await this.closeTransport() } - private async call(method: string, args: unknown[]): Promise { + private async call(method: string, args: unknown[], options?: { scheduleIdle?: boolean }): Promise { const transport = this.ensureTransport() this.pendingRequests++ this.clearIdleTimer() @@ -187,7 +216,21 @@ export class SemanticIndexWorkerClient implements SemanticIndexRuntime { return await transport.request(method, args) } finally { this.pendingRequests-- - this.scheduleIdleCloseIfNeeded() + if (options?.scheduleIdle !== false) this.scheduleIdleCloseIfNeeded() + } + } + + private async callProbe( + method: string, + args: unknown[], + fallback: T, + options?: { scheduleIdle?: boolean } + ): Promise { + try { + return await this.call(method, args, options) + } catch { + await this.closeFailedProbeTransport() + return fallback } } @@ -198,12 +241,50 @@ export class SemanticIndexWorkerClient implements SemanticIndexRuntime { return this.transport } - private updateActiveBuildFromStatuses(statuses: SemanticIndexSessionStatus[]): void { - this.hasActiveBuild = statuses.some((status) => status.running || status.queued || status.indexStatus === 'running') + private canRunFromLocalConfig(): boolean { + const config = this.configStore.get() + if (!this.configStore.canRun()) return false + if (config.mode !== 'api') return true + if (isKeylessSemanticIndexApiBaseUrl(config.api?.baseUrl)) return true + return resolveSemanticIndexApiKeySet(config, this.resolveApiKey) + } + + private updateActiveBuildFromCompleteSnapshot(statuses: SemanticIndexSessionStatus[]): void { + this.hasUnknownActiveBuild = false + this.activeBuildSessionIds = new Set( + statuses.filter((status) => this.isActiveBuildStatus(status)).map((status) => status.sessionId) + ) + } + + private updateActiveBuildFromPartialSnapshot( + requestedSessionIds: string[], + statuses: SemanticIndexSessionStatus[] + ): void { + const bySessionId = new Map(statuses.map((status) => [status.sessionId, status])) + for (const sessionId of requestedSessionIds) { + const status = bySessionId.get(sessionId) + if (status && this.isActiveBuildStatus(status)) this.activeBuildSessionIds.add(sessionId) + else this.activeBuildSessionIds.delete(sessionId) + } + } + + private isActiveBuildStatus(status: SemanticIndexSessionStatus): boolean { + return status.running || status.queued || status.indexStatus === 'running' + } + + private clearActiveBuildTracking(): void { + this.hasUnknownActiveBuild = false + this.activeBuildSessionIds.clear() } private scheduleIdleCloseIfNeeded(): void { - if (!this.transport || this.pendingRequests > 0 || this.hasActiveBuild) return + if ( + !this.transport || + this.pendingRequests > 0 || + this.hasUnknownActiveBuild || + this.activeBuildSessionIds.size > 0 + ) + return this.clearIdleTimer() this.idleTimer = this.timers.setTimeout(() => { void this.closeTransport().catch((err) => { @@ -222,9 +303,18 @@ export class SemanticIndexWorkerClient implements SemanticIndexRuntime { private async closeTransport(): Promise { const transport = this.transport this.transport = null - this.hasActiveBuild = false + this.clearActiveBuildTracking() if (transport) await transport.close() } + + private async closeFailedProbeTransport(): Promise { + this.clearIdleTimer() + try { + await this.closeTransport() + } catch { + // 语义索引探针失败只降级为不可用，不能打断普通 AI 对话流程。 + } + } } export function createSemanticIndexWorkerClient(options: SemanticIndexWorkerClientOptions): SemanticIndexWorkerClient { diff --git a/packages/sync/src/pull-engine.test.ts b/packages/sync/src/pull-engine.test.ts index 0f7671fa..4ecc2fb8 100644 --- a/packages/sync/src/pull-engine.test.ts +++ b/packages/sync/src/pull-engine.test.ts @@ -292,4 +292,107 @@ describe('PullEngine', () => { assert.equal(result.success, true) assert.equal(sessionUpdates.at(-1)?.updates.lastPullAt, 1940) }) + + it('persists terminal retry nextSince when it is newer than imported messages', async () => { + const session = createSession() + session.lastPullAt = 0 + const dataSource = createDataSource() + dataSource.sessions = [session] + const emptyInitialPage = writeTempJson({ + chatlab: { version: '0.0.2', exportedAt: 100 }, + meta: { name: 'Test Session', platform: 'test', type: 'group' }, + members: [], + messages: [], + sync: { hasMore: false, nextSince: 100 }, + }) + const terminalRetryPage = writeTempJson({ + chatlab: { version: '0.0.2', exportedAt: 100 }, + meta: { name: 'Test Session', platform: 'test', type: 'group' }, + members: [{ platformId: 'u1', accountName: 'Alice' }], + messages: [{ sender: 'u1', timestamp: 2000, type: 0, content: 'late retry message' }], + sync: { hasMore: false, nextSince: 5000 }, + }) + const sessionUpdates: Array<{ sessionId: string; updates: Partial }> = [] + const engine = createEngine({ + files: [emptyInitialPage, terminalRetryPage], + dataSource, + sessionUpdates, + importResult: { + success: true, + newMessageCount: 1, + sessionId: session.targetSessionId, + }, + }) + + const result = await withImmediateTimers(() => engine.executePullSession(dataSource.id, dataSource, session)) + + assert.equal(result.success, true) + assert.equal(sessionUpdates.at(-1)?.updates.lastPullAt, 4940) + }) + + it('persists terminal retry nextSince from empty retry pages', async () => { + const session = createSession() + session.lastPullAt = 0 + const dataSource = createDataSource() + dataSource.sessions = [session] + const emptyPages = Array.from({ length: 4 }, () => + writeTempJson({ + chatlab: { version: '0.0.2', exportedAt: 100 }, + meta: { name: 'Test Session', platform: 'test', type: 'group' }, + members: [], + messages: [], + sync: { hasMore: false, nextSince: 5000 }, + }) + ) + const sessionUpdates: Array<{ sessionId: string; updates: Partial }> = [] + const engine = createEngine({ + files: emptyPages, + dataSource, + sessionUpdates, + importResult: { + success: true, + newMessageCount: 0, + sessionId: session.targetSessionId, + }, + }) + + const result = await withImmediateTimers(() => engine.executePullSession(dataSource.id, dataSource, session)) + + assert.equal(result.success, true) + assert.equal(result.newMessageCount, 0) + assert.equal(sessionUpdates.at(-1)?.updates.lastPullAt, 4940) + }) + + it('keeps the pull cursor stable when a successful pull returns no new cursor', async () => { + const session = createSession() + session.lastPullAt = 100 + const dataSource = createDataSource() + dataSource.sessions = [session] + const emptyPages = Array.from({ length: 4 }, () => + writeTempJson({ + chatlab: { version: '0.0.2', exportedAt: 100 }, + meta: { name: 'Test Session', platform: 'test', type: 'group' }, + members: [], + messages: [], + sync: { hasMore: false }, + }) + ) + const sessionUpdates: Array<{ sessionId: string; updates: Partial }> = [] + const engine = createEngine({ + files: emptyPages, + dataSource, + sessionUpdates, + importResult: { + success: true, + newMessageCount: 0, + sessionId: session.targetSessionId, + }, + }) + + const result = await withImmediateTimers(() => engine.executePullSession(dataSource.id, dataSource, session)) + + assert.equal(result.success, true) + assert.equal(result.newMessageCount, 0) + assert.equal(sessionUpdates.at(-1)?.updates.lastPullAt, 100) + }) }) diff --git a/packages/sync/src/pull-engine.ts b/packages/sync/src/pull-engine.ts index bd2a5f1e..8029b2fe 100644 --- a/packages/sync/src/pull-engine.ts +++ b/packages/sync/src/pull-engine.ts @@ -272,6 +272,8 @@ export class PullEngine { this.logger.info(`[Pull] "${sess.name}" retry ${ri + 1}: fetched ${retryStat.size} bytes`) const retrySync = parseSyncFromFile(retryFile) if (retryStat.size < 1024 && !fileContainsMessages(retryFile) && retrySync?.hasMore === false) { + // 空 retry 页也可能只返回服务端 watermark，跳过导入前仍要推进保存游标。 + if (retrySync.nextSince !== undefined) nextPullSince = Math.max(nextPullSince, retrySync.nextSince) cleanupTempFile(retryFile) continue } @@ -325,6 +327,8 @@ export class PullEngine { } totalNewMessages += retryResult.newMessageCount if (retryMaxTs !== null) nextPullSince = Math.max(nextPullSince, retryMaxTs) + // retry 终止页也可能携带服务端 watermark，必须保存，否则下次会重复拉取尾部窗口。 + if (retrySync?.nextSince !== undefined) nextPullSince = Math.max(nextPullSince, retrySync.nextSince) this.progressMap.set(sess.id, { sessionId: sess.id, sessionName: sess.name, @@ -335,7 +339,6 @@ export class PullEngine { if (retrySync?.hasMore && retrySync.nextSince !== undefined) { since = retrySync.nextSince offset = undefined - nextPullSince = Math.max(nextPullSince, retrySync.nextSince) } else if (retrySync?.hasMore && retrySync.nextOffset !== undefined) { offset = retrySync.nextOffset } @@ -436,8 +439,10 @@ export class PullEngine { this.logger.warn(`[Pull] "${sess.name}" reached page limit (${MAX_PAGES_PER_PULL}), data may be incomplete`) } + // 保留 overlap 窗口，但成功拉取未观察到更新游标时不能把已保存游标向后移动。 + const savedLastPullAt = Math.max(sess.lastPullAt, Math.max(0, nextPullSince - PULL_OVERLAP_SECONDS)) this.dsManager.updateSession(sourceId, sess.id, { - lastPullAt: Math.max(0, nextPullSince - PULL_OVERLAP_SECONDS), + lastPullAt: savedLastPullAt, lastStatus: 'success', lastNewMessages: totalNewMessages, lastError: '',