fix(ai): bound cross-chat candidate discovery

Make unscoped candidate preparation honor cancellation and wall-time limits before message scanning. Enumerate database files lazily, retain only the newest eligible sessions, and resolve owners without loading full member statistics.
This commit is contained in:
digua committed 2026-08-22 17:09:23 +08:00
1 parent d1a6c711fe
commit 5d5355d4f1
5 files changed
+124 -26

No files matched your search

+10 -4
View File
@@ -186,16 +186,22 @@ export class DatabaseManager {
* 列举数据库目录下的所有聊天会话 ID
*/
listSessionIds(): string[] {
this.assertCompatible()
const dbDir = this.pathProvider.getDatabaseDir()
return listDatabaseCandidateIds(dbDir).filter((id) => {
return this.listSessionCandidateIds().filter((id) => {
const db = this.open(id)
if (!db) return false
return isChatSessionDb(db)
})
}
/**
* 仅列出数据库文件候选,不同步打开所有数据库。
* 需要逐库检查、取消和时间预算的调用方应使用此入口后自行验证。
*/
listSessionCandidateIds(): string[] {
this.assertCompatible()
return listDatabaseCandidateIds(this.pathProvider.getDatabaseDir())
}
/**
* 只读列举聊天会话 ID,不触发数据库迁移或兼容门禁提升。
*/
@@ -12,6 +12,7 @@ import type { SessionRuntimeAdapter } from './types'
export function createDatabaseManagerAdapter(dbManager: DatabaseManager): SessionRuntimeAdapter {
return {
listSessionIds: () => dbManager.listSessionIds(),
listSessionCandidateIds: () => dbManager.listSessionCandidateIds(),
openReadonly: (sessionId) => dbManager.open(sessionId),
openWritable: (sessionId) => dbManager.openWritable(sessionId),
closeSession: (sessionId) => dbManager.close(sessionId),
@@ -9,6 +9,8 @@ import type { DatabaseAdapter } from '@openchatlab/core'
export interface SessionRuntimeAdapter {
listSessionIds(): string[]
/** List database file candidates without opening every database. */
listSessionCandidateIds?(): string[]
openReadonly(sessionId: string): DatabaseAdapter | null
openWritable(sessionId: string): DatabaseAdapter | null
closeSession(sessionId: string): void
@@ -46,6 +46,7 @@ class TestEnvironment {
}
this.adapter = {
listSessionIds: () => [...this.dbPaths.keys()],
listSessionCandidateIds: () => [...this.dbPaths.keys()],
openReadonly: (sessionId) => open(sessionId, true),
openWritable: (sessionId) => open(sessionId, false),
closeSession: () => {},
@@ -582,14 +583,24 @@ test('search honors interruption between session scans', async () => {
}
})
test('search stops between sessions when the wall-time budget is exhausted', async () => {
test('search stops during candidate preparation when the wall-time budget is exhausted', async () => {
const { env, contactsService } = createFixture()
try {
const timestamps = [0, 0, 9_000]
let now = 0
const openedSessions: string[] = []
const adapter: SessionRuntimeAdapter = {
...env.adapter,
openReadonly: (sessionId) => {
openedSessions.push(sessionId)
const db = env.adapter.openReadonly(sessionId)
now += 9_000
return db
},
}
const service = createCrossChatAnalysisService({
adapter: env.adapter,
adapter,
contactsService,
now: () => timestamps.shift() ?? 9_000,
now: () => now,
})
const result = await service.searchMessages({
keywords: ['project alpha'],
@@ -598,7 +609,9 @@ test('search stops between sessions when the wall-time budget is exhausted', asy
maxWallTimeMs: 8_000,
})
assert.equal(result.coverage.scannedSessions, 1)
assert.equal(openedSessions.length, 1)
assert.equal(result.coverage.candidateSessions, 3)
assert.equal(result.coverage.scannedSessions, 0)
assert.equal(result.coverage.truncated, true)
assert.ok(result.coverage.truncatedReasons.includes('time_budget'))
} finally {
@@ -606,6 +619,31 @@ test('search stops between sessions when the wall-time budget is exhausted', asy
}
})
test('search honors interruption while resolving unscoped candidates', async () => {
const { env, contactsService } = createFixture()
try {
const openedSessions: string[] = []
const adapter: SessionRuntimeAdapter = {
...env.adapter,
openReadonly: (sessionId) => {
openedSessions.push(sessionId)
return env.adapter.openReadonly(sessionId)
},
}
const controller = new AbortController()
const service = createCrossChatAnalysisService({ adapter, contactsService })
const search = service.searchMessages({ keywords: ['project'], maxSessions: 3 }, { signal: controller.signal })
queueMicrotask(() => controller.abort())
await assert.rejects(search, { name: 'AbortError' })
assert.ok(openedSessions.length > 0)
assert.ok(openedSessions.length < adapter.listSessionCandidateIds!().length)
} finally {
env.cleanup()
}
})
test('overview preserves separate member scopes for contacts in the same group', async () => {
const { env, contactsService } = createFixture()
try {
@@ -4,6 +4,7 @@ import {
getSearchMessageContext as getCoreSearchMessageContext,
getSessionMeta,
getSessionOverview,
resolveOwnerMember,
searchMessagesByKeywords,
type DatabaseAdapter,
} from '@openchatlab/core'
@@ -255,15 +256,19 @@ class DefaultCrossChatAnalysisService implements CrossChatAnalysisService {
const maxWallTimeMs = clampInteger(request.maxWallTimeMs, DEFAULT_MAX_WALL_TIME_MS, 1, MAX_MAX_WALL_TIME_MS)
const startedAt = this.now()
const failedSessionIds = new Set<string>()
const { candidates, candidateSessionCount } = this.resolveSearchCandidates(
const { candidates, candidateSessionCount, eligibleCandidateCount, timedOut } = await this.resolveSearchCandidates(
request.scopes,
failedSessionIds,
sender,
ownerResolution
ownerResolution,
maxSessions,
startedAt + maxWallTimeMs,
options.signal
)
const selected = candidates.slice(0, maxSessions)
const selected = candidates
const truncatedReasons = new Set<CrossChatTruncationReason>()
if (candidates.length > selected.length) truncatedReasons.add('session_budget')
if (eligibleCandidateCount > selected.length) truncatedReasons.add('session_budget')
if (timedOut) truncatedReasons.add('time_budget')
const messages: CrossChatMessageSource[] = []
let totalMatches = 0
@@ -424,7 +429,7 @@ class DefaultCrossChatAnalysisService implements CrossChatAnalysisService {
}
}
private resolveSearchCandidates(
private async resolveSearchCandidates(
scopes: CrossChatSearchScope[] | undefined,
failedSessionIds: Set<string>,
sender: 'all' | 'owner',
@@ -432,47 +437,76 @@ class DefaultCrossChatAnalysisService implements CrossChatAnalysisService {
resolvedSessions: number
missingOwnerSessions: number
unresolvedOwnerSessions: number
}
): { candidates: SearchCandidate[]; candidateSessionCount: number } {
},
maxSessions = DEFAULT_MAX_SESSIONS,
deadline = Number.POSITIVE_INFINITY,
signal?: AbortSignal
): Promise<{
candidates: SearchCandidate[]
candidateSessionCount: number
eligibleCandidateCount: number
timedOut: boolean
}> {
throwIfAborted(signal)
const normalizedScopes: CrossChatSearchScope[] = scopes
? normalizeScopes(scopes)
: this.deps.adapter.listSessionIds().map((sessionId) => ({ sessionId }))
: (this.deps.adapter.listSessionCandidateIds?.() ?? this.deps.adapter.listSessionIds()).map((sessionId) => ({
sessionId,
}))
const candidates: SearchCandidate[] = []
let eligibleCandidateCount = 0
let timedOut = false
for (const scope of normalizedScopes) {
throwIfAborted(signal)
if (this.now() >= deadline) {
timedOut = true
break
}
const descriptor = this.tryGetSessionDescriptor(scope.sessionId)
if (!descriptor) {
failedSessionIds.add(scope.sessionId)
await yieldToEventLoop()
continue
}
if (this.now() >= deadline) {
timedOut = true
break
}
if (sender === 'owner') {
const db = this.deps.adapter.openReadonly(scope.sessionId)
if (!db) {
failedSessionIds.add(scope.sessionId)
await yieldToEventLoop()
continue
}
const meta = getSessionMeta(db)
if (!meta?.ownerId?.trim()) {
if (ownerResolution) ownerResolution.missingOwnerSessions++
await yieldToEventLoop()
continue
}
const owner = getMembers(db).find((member) => member.platformId === meta.ownerId)
const owner = resolveOwnerMember(db)
if (!owner) {
if (ownerResolution) ownerResolution.unresolvedOwnerSessions++
await yieldToEventLoop()
continue
}
if (ownerResolution) ownerResolution.resolvedSessions++
candidates.push({ descriptor, memberIds: [owner.id] })
continue
eligibleCandidateCount++
addBoundedSearchCandidate(candidates, { descriptor, memberIds: [owner.id] }, maxSessions)
} else {
eligibleCandidateCount++
addBoundedSearchCandidate(candidates, { descriptor, memberIds: scope.memberIds }, maxSessions)
}
candidates.push({ descriptor, memberIds: scope.memberIds })
await yieldToEventLoop()
}
throwIfAborted(signal)
if (!timedOut && this.now() >= deadline) timedOut = true
return {
candidates: candidates.sort(
(left, right) =>
(right.descriptor.lastMessageTs ?? Number.NEGATIVE_INFINITY) -
(left.descriptor.lastMessageTs ?? Number.NEGATIVE_INFINITY)
),
candidates,
candidateSessionCount: normalizedScopes.length,
eligibleCandidateCount,
timedOut,
}
}
@@ -506,6 +540,23 @@ interface SearchCandidate {
memberIds?: number[]
}
function addBoundedSearchCandidate(
candidates: SearchCandidate[],
candidate: SearchCandidate,
maxSessions: number
): void {
candidates.push(candidate)
candidates.sort(compareSearchCandidates)
if (candidates.length > maxSessions) candidates.length = maxSessions
}
function compareSearchCandidates(left: SearchCandidate, right: SearchCandidate): number {
return (
(right.descriptor.lastMessageTs ?? Number.NEGATIVE_INFINITY) -
(left.descriptor.lastMessageTs ?? Number.NEGATIVE_INFINITY)
)
}
function getSessionDescriptor(sessionId: string, db: DatabaseAdapter): CrossChatSessionDescriptor | null {
const meta = getSessionMeta(db)
if (!meta || (meta.type !== ChatType.PRIVATE && meta.type !== ChatType.GROUP)) return null