feat(ai): add cross-chat analysis service

This commit is contained in:
digua committed 2026-08-22 07:43:31 +08:00
1 parent 71a5636392
commit 831861a993
9 files changed
+1116

No files matched your search

@@ -177,6 +177,18 @@ describe('search extensions (in-memory SQLite)', () => {
[7, 6]
)
})
it('filters multiple senders in one query', () => {
raw.prepare("INSERT INTO member (id, platform_id, account_name) VALUES (3, 'u3', 'Carol')").run()
raw
.prepare('INSERT INTO message (id, sender_id, ts, type, content) VALUES (?, ?, ?, ?, ?)')
.run(6, 3, 6000, 0, 'unselected sender')
const result = searchMessagesByKeywords(db, [], { senderIds: [1, 2], sort: 'asc' })
assert.deepEqual(
result.messages.map((message) => message.id),
[1, 2, 3, 4, 5]
)
})
})
describe('getConversationBetween pagination and blacklist (in-memory SQLite)', () => {
@@ -212,6 +212,7 @@ export function searchMessagesByKeywords(
startTs?: number
endTs?: number
senderId?: number
senderIds?: number[]
limit?: number
offset?: number
/** Keyword join mode: 'any' (OR, default) or 'all' (AND). */
@@ -231,6 +232,7 @@ export function searchMessagesByKeywords(
startTs: options?.startTs,
endTs: options?.endTs,
senderId: options?.senderId,
senderIds: options?.senderIds,
keywords: cleaned.length > 0 ? cleaned : undefined,
matchMode: options?.matchMode,
excludeKeywords: options?.excludeKeywords,
+9
View File
@@ -129,6 +129,7 @@ export function buildMsgConditions(options?: {
startTs?: number
endTs?: number
senderId?: number
senderIds?: number[]
memberId?: number | null
keywords?: string[]
/** Keyword join mode: 'any' (OR, default) or 'all' (AND). */
@@ -156,6 +157,14 @@ export function buildMsgConditions(options?: {
if (options?.senderId != null) {
conds.push('msg.sender_id = ?')
params.push(options.senderId)
} else if (options?.senderIds) {
const senderIds = [...new Set(options.senderIds)]
if (senderIds.length === 0) {
conds.push('1 = 0')
} else {
conds.push(`msg.sender_id IN (${senderIds.map(() => '?').join(', ')})`)
params.push(...senderIds)
}
}
if (options?.memberId != null) {
conds.push('msg.sender_id = ?')
+22
View File
@@ -454,6 +454,7 @@ export {
type ImportDemoSessionsOptions,
} from './services/demo-import'
export * as contactsService from './services/contacts'
export * as crossChatAnalysisService from './services/cross-chat-analysis'
export * as peopleRelationshipsService from './services/people/relationships'
export * as globalInsightService from './services/global-insight'
// Semantic index (Phase 1 vector search) — independent of legacy ai/rag
@@ -493,6 +494,7 @@ export {
TIME_INVESTMENT_ALGORITHM_VERSION,
MergeSessionCache,
createContactsService,
createCrossChatAnalysisService,
createPeopleRelationshipsService,
createGlobalInsightService,
assertValidTopicDayKey,
@@ -526,6 +528,26 @@ export type {
ContactsService,
ContactsServiceDeps,
ContactsServiceOptions,
CrossChatAnalysisService,
CrossChatAnalysisServiceDeps,
CrossChatEntityResolution,
CrossChatMessageContextRequest,
CrossChatMessageContextResult,
CrossChatMessageSource,
CrossChatOperationOptions,
CrossChatOverviewItem,
CrossChatOverviewRequest,
CrossChatOverviewResult,
CrossChatResolvedContact,
CrossChatResolvedContactSession,
CrossChatResolvedSession,
CrossChatSearchProgress,
CrossChatSearchRequest,
CrossChatSearchResult,
CrossChatSearchScope,
CrossChatSessionDescriptor,
CrossChatTruncationReason,
CrossChatUnresolvedEntity,
PeopleRelationshipsService,
PeopleRelationshipsServiceDeps,
PeopleRelationshipsServiceOptions,
@@ -0,0 +1,22 @@
export { createCrossChatAnalysisService } from './service'
export type { CrossChatAnalysisService, CrossChatAnalysisServiceDeps } from './service'
export type {
CrossChatEntityResolution,
CrossChatMessageContextRequest,
CrossChatMessageContextResult,
CrossChatMessageSource,
CrossChatOperationOptions,
CrossChatOverviewItem,
CrossChatOverviewRequest,
CrossChatOverviewResult,
CrossChatResolvedContact,
CrossChatResolvedContactSession,
CrossChatResolvedSession,
CrossChatSearchProgress,
CrossChatSearchRequest,
CrossChatSearchResult,
CrossChatSearchScope,
CrossChatSessionDescriptor,
CrossChatTruncationReason,
CrossChatUnresolvedEntity,
} from './types'
@@ -0,0 +1,385 @@
import assert from 'node:assert/strict'
import fs from 'node:fs'
import os from 'node:os'
import path from 'node:path'
import test from 'node:test'
import { CHAT_DB_SCHEMA } from '@openchatlab/core'
import type { DatabaseAdapter } from '@openchatlab/core'
import type { ContactDetailResponse, ContactItem } from '@openchatlab/shared-types'
import { openBetterSqliteDatabase } from '../../better-sqlite3-adapter'
import type { ContactsService } from '../contacts'
import type { SessionRuntimeAdapter } from '../adapters'
import { createCrossChatAnalysisService } from './service'
const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node')
interface SeedSession {
id: string
name: string
type: 'private' | 'group'
members: Array<{ id: number; platformId: string; name: string }>
messages: Array<{ id: number; senderId: number; ts: number; content: string }>
}
class TestEnvironment {
readonly dir: string
readonly adapter: SessionRuntimeAdapter
private readonly dbPaths = new Map<string, string>()
private readonly openDatabases: DatabaseAdapter[] = []
constructor() {
const baseDir = process.env.CHATLAB_TEST_TMPDIR ?? (fs.existsSync('/private/tmp') ? '/private/tmp' : os.tmpdir())
this.dir = fs.mkdtempSync(path.join(baseDir, 'chatlab-cross-chat-analysis-'))
const open = (sessionId: string, readonly: boolean): DatabaseAdapter | null => {
const dbPath = this.dbPaths.get(sessionId)
if (!dbPath) return null
const db = openBetterSqliteDatabase(dbPath, { readonly, nativeBinding })
this.openDatabases.push(db)
return db
}
this.adapter = {
listSessionIds: () => [...this.dbPaths.keys()],
openReadonly: (sessionId) => open(sessionId, true),
openWritable: (sessionId) => open(sessionId, false),
closeSession: () => {},
getDbPath: (sessionId) => this.dbPaths.get(sessionId) ?? '',
deleteSessionFile: () => false,
ensureReadonly: (sessionId) => {
const db = open(sessionId, true)
if (!db) throw Object.assign(new Error(`Session not found: ${sessionId}`), { statusCode: 404 })
return db
},
ensureWritable: (sessionId) => {
const db = open(sessionId, false)
if (!db) throw Object.assign(new Error(`Session not found: ${sessionId}`), { statusCode: 404 })
return db
},
}
}
seed(session: SeedSession): void {
const dbPath = path.join(this.dir, `${session.id}.db`)
const db = openBetterSqliteDatabase(dbPath, { nativeBinding })
db.exec(CHAT_DB_SCHEMA)
db.prepare('INSERT INTO meta (name, platform, type, imported_at) VALUES (?, ?, ?, ?)').run(
session.name,
'test',
session.type,
1780000000
)
for (const member of session.members) {
db.prepare('INSERT INTO member (id, platform_id, account_name) VALUES (?, ?, ?)').run(
member.id,
member.platformId,
member.name
)
}
for (const message of session.messages) {
db.prepare(
'INSERT INTO message (id, sender_id, ts, type, content, platform_message_id) VALUES (?, ?, ?, 0, ?, ?)'
).run(message.id, message.senderId, message.ts, message.content, `${session.id}-${message.id}`)
}
db.close()
this.dbPaths.set(session.id, dbPath)
}
cleanup(): void {
for (const db of this.openDatabases) {
try {
db.close()
} catch {
// A test may have already closed the handle.
}
}
fs.rmSync(this.dir, { recursive: true, force: true })
}
}
function contact(
overrides: Partial<ContactItem> & Pick<ContactItem, 'key' | 'platformId' | 'displayName'>
): ContactItem {
return {
key: overrides.key,
platform: overrides.platform ?? 'test',
platformId: overrides.platformId,
sessionScoped: overrides.sessionScoped ?? false,
sessionId: overrides.sessionId,
displayName: overrides.displayName,
aliases: overrides.aliases ?? [],
avatar: null,
isFriend: true,
pool: 'friend',
score: 1,
scoreBreakdown: {},
sourceSessions: overrides.sourceSessions ?? [],
searchText: '',
lastInteractionTs: overrides.lastInteractionTs ?? null,
}
}
function detail(
item: ContactItem | null,
cacheStatus: ContactDetailResponse['cache']['status'] = 'fresh'
): ContactDetailResponse {
return {
contact: item,
cache: { status: cacheStatus, computedAt: 1780000000 },
timeRange: { preset: 'all', anchorTs: null, startTs: null },
algorithmVersion: 'test',
}
}
function createFixture(): {
env: TestEnvironment
contactsService: Pick<ContactsService, 'getContactDetail'>
} {
const env = new TestEnvironment()
env.seed({
id: 'private-alice',
name: 'Alice private',
type: 'private',
members: [
{ id: 1, platformId: 'owner', name: 'Me' },
{ id: 10, platformId: 'alice', name: 'Alice' },
],
messages: [
{ id: 1, senderId: 10, ts: 100, content: 'project alpha early note' },
{ id: 2, senderId: 1, ts: 110, content: 'project alpha response' },
],
})
env.seed({
id: 'group-work',
name: 'Work group',
type: 'group',
members: [
{ id: 1, platformId: 'owner', name: 'Me' },
{ id: 20, platformId: 'alice', name: 'Alice' },
{ id: 21, platformId: 'bob', name: 'Bob' },
],
messages: [
{ id: 1, senderId: 20, ts: 300, content: 'project alpha latest decision' },
{ id: 2, senderId: 21, ts: 310, content: 'project alpha unrelated sender' },
{ id: 3, senderId: 20, ts: 320, content: 'follow-up context' },
],
})
env.seed({
id: 'group-other',
name: 'Other group',
type: 'group',
members: [{ id: 30, platformId: 'alice-other', name: 'Alice' }],
messages: [{ id: 1, senderId: 30, ts: 200, content: 'project alpha from another Alice' }],
})
const contacts = new Map<string, ContactDetailResponse>([
[
'test:alice',
detail(
contact({
key: 'test:alice',
platformId: 'alice',
displayName: 'Alice',
sourceSessions: [
{ id: 'private-alice', name: 'Alice private', platform: 'test', type: 'private' },
{ id: 'group-work', name: 'Work group', platform: 'test', type: 'group' },
],
})
),
],
[
'test:group-other:alice-other',
detail(
contact({
key: 'test:group-other:alice-other',
platformId: 'alice-other',
displayName: 'Alice',
sessionScoped: true,
sessionId: 'group-other',
sourceSessions: [{ id: 'group-other', name: 'Other group', platform: 'test', type: 'group' }],
})
),
],
])
return {
env,
contactsService: {
getContactDetail: (key) => contacts.get(key) ?? detail(null),
},
}
}
test('entity resolution uses contact keys and resolves per-session member ids without merging display names', () => {
const { env, contactsService } = createFixture()
try {
const service = createCrossChatAnalysisService({ adapter: env.adapter, contactsService })
const result = service.resolveEntities([
{ type: 'contact', contactKey: 'test:alice', displayName: 'Alice' },
{ type: 'contact', contactKey: 'test:group-other:alice-other', displayName: 'Alice' },
])
assert.deepEqual(
result.contacts.map((item) => [
item.ref.contactKey,
item.sessions.map((session) => [session.sessionId, session.memberId]),
]),
[
[
'test:alice',
[
['private-alice', 10],
['group-work', 20],
],
],
['test:group-other:alice-other', [['group-other', 30]]],
]
)
assert.equal(result.coverage.resolvedEntities, 2)
assert.equal(result.coverage.resolvedSessions, 3)
} finally {
env.cleanup()
}
})
test('scoped search filters by resolved member ids and keeps compound evidence identity', async () => {
const { env, contactsService } = createFixture()
try {
const service = createCrossChatAnalysisService({ adapter: env.adapter, contactsService })
const result = await service.searchMessages({
keywords: ['project alpha'],
scopes: [
{ sessionId: 'private-alice', memberIds: [10] },
{ sessionId: 'group-work', memberIds: [20] },
],
maxEvidence: 10,
})
assert.deepEqual(
result.messages.map((message) => [message.sessionId, message.messageId, message.senderId]),
[
['group-work', 1, 20],
['private-alice', 1, 10],
]
)
assert.equal(result.coverage.truncated, false)
const context = service.getMessageContext({ sessionId: 'group-work', messageId: 1, contextSize: 1 })
assert.deepEqual(
context.messages.map((message) => [message.sessionId, message.messageId]),
[
['group-work', 1],
['group-work', 2],
]
)
} finally {
env.cleanup()
}
})
test('global search scans recent sessions first and reports budget truncation', async () => {
const { env, contactsService } = createFixture()
try {
const service = createCrossChatAnalysisService({ adapter: env.adapter, contactsService })
const progress: string[] = []
const result = await service.searchMessages(
{ keywords: ['project alpha'], maxSessions: 1, maxEvidence: 10 },
{ onProgress: (item) => item.currentSessionId && progress.push(item.currentSessionId) }
)
assert.deepEqual(progress, ['group-work'])
assert.deepEqual(
result.messages.map((message) => message.sessionId),
['group-work', 'group-work']
)
assert.equal(result.coverage.candidateSessions, 3)
assert.equal(result.coverage.scannedSessions, 1)
assert.equal(result.coverage.truncated, true)
assert.ok(result.coverage.truncatedReasons.includes('session_budget'))
} finally {
env.cleanup()
}
})
test('search rejects empty keywords and honors interruption before scanning', async () => {
const { env, contactsService } = createFixture()
try {
const service = createCrossChatAnalysisService({ adapter: env.adapter, contactsService })
await assert.rejects(() => service.searchMessages({ keywords: [] }), /keyword/i)
const controller = new AbortController()
controller.abort()
await assert.rejects(() => service.searchMessages({ keywords: ['project'] }, { signal: controller.signal }), {
name: 'AbortError',
})
} finally {
env.cleanup()
}
})
test('search honors interruption between session scans', async () => {
const { env, contactsService } = createFixture()
try {
const service = createCrossChatAnalysisService({ adapter: env.adapter, contactsService })
const controller = new AbortController()
await assert.rejects(
() =>
service.searchMessages(
{ keywords: ['project'], maxSessions: 3 },
{
signal: controller.signal,
onProgress: (progress) => {
if (progress.currentSessionId === 'group-other') controller.abort()
},
}
),
{ name: 'AbortError' }
)
} finally {
env.cleanup()
}
})
test('search stops between sessions when the wall-time budget is exhausted', async () => {
const { env, contactsService } = createFixture()
try {
const timestamps = [0, 0, 9_000]
const service = createCrossChatAnalysisService({
adapter: env.adapter,
contactsService,
now: () => timestamps.shift() ?? 9_000,
})
const result = await service.searchMessages({
keywords: ['project alpha'],
maxSessions: 3,
maxEvidence: 10,
maxWallTimeMs: 8_000,
})
assert.equal(result.coverage.scannedSessions, 1)
assert.equal(result.coverage.truncated, true)
assert.ok(result.coverage.truncatedReasons.includes('time_budget'))
} finally {
env.cleanup()
}
})
test('overview returns separate scoped totals instead of combining unrelated senders', async () => {
const { env, contactsService } = createFixture()
try {
const service = createCrossChatAnalysisService({ adapter: env.adapter, contactsService })
const result = await service.getOverview({
scopes: [
{ sessionId: 'group-work', memberIds: [20], label: 'Alice in Work group' },
{ sessionId: 'group-other', memberIds: [30], label: 'Other Alice' },
],
})
assert.deepEqual(
result.items.map((item) => [item.label, item.totalMessages, item.firstMessageTs, item.lastMessageTs]),
[
['Alice in Work group', 2, 300, 320],
['Other Alice', 1, 200, 200],
]
)
} finally {
env.cleanup()
}
})
@@ -0,0 +1,499 @@
import {
getMembers,
getMessageContext as getCoreMessageContext,
getRecentMessages,
getSessionMeta,
getSessionOverview,
searchMessagesByKeywords,
type DatabaseAdapter,
} from '@openchatlab/core'
import { ChatType } from '@openchatlab/shared-types'
import type { AIEntityRef } from '../../ai/chats'
import { appLogger } from '../../logging/app-logger'
import type { SessionRuntimeAdapter } from '../adapters'
import type { ContactsService } from '../contacts'
import type {
CrossChatEntityResolution,
CrossChatMessageContextRequest,
CrossChatMessageContextResult,
CrossChatMessageSource,
CrossChatOperationOptions,
CrossChatOverviewItem,
CrossChatOverviewRequest,
CrossChatOverviewResult,
CrossChatResolvedContact,
CrossChatResolvedSession,
CrossChatSearchRequest,
CrossChatSearchResult,
CrossChatSearchScope,
CrossChatSessionDescriptor,
CrossChatTruncationReason,
CrossChatUnresolvedEntity,
} from './types'
const DEFAULT_MAX_SESSIONS = 24
const MAX_MAX_SESSIONS = 100
const DEFAULT_MAX_EVIDENCE = 80
const MAX_MAX_EVIDENCE = 200
const DEFAULT_MAX_WALL_TIME_MS = 8_000
const MAX_MAX_WALL_TIME_MS = 30_000
const DEFAULT_CONTEXT_SIZE = 10
const MAX_CONTEXT_SIZE = 50
export interface CrossChatAnalysisServiceDeps {
adapter: SessionRuntimeAdapter
contactsService: Pick<ContactsService, 'getContactDetail'>
now?: () => number
}
export interface CrossChatAnalysisService {
resolveEntities(refs: AIEntityRef[]): CrossChatEntityResolution
searchMessages(request: CrossChatSearchRequest, options?: CrossChatOperationOptions): Promise<CrossChatSearchResult>
getMessageContext(request: CrossChatMessageContextRequest): CrossChatMessageContextResult
getOverview(request: CrossChatOverviewRequest, options?: CrossChatOperationOptions): Promise<CrossChatOverviewResult>
}
export function createCrossChatAnalysisService(deps: CrossChatAnalysisServiceDeps): CrossChatAnalysisService {
return new DefaultCrossChatAnalysisService(deps)
}
class DefaultCrossChatAnalysisService implements CrossChatAnalysisService {
constructor(private readonly deps: CrossChatAnalysisServiceDeps) {}
resolveEntities(refs: AIEntityRef[]): CrossChatEntityResolution {
const contacts: CrossChatResolvedContact[] = []
const sessions: CrossChatResolvedSession[] = []
const unresolved: CrossChatUnresolvedEntity[] = []
const candidateSessionIds = new Set<string>()
const resolvedSessionIds = new Set<string>()
const failedSessionIds = new Set<string>()
for (const ref of refs) {
if (ref.type === 'session') {
candidateSessionIds.add(ref.sessionId)
const descriptor = this.tryGetSessionDescriptor(ref.sessionId)
if (descriptor) {
resolvedSessionIds.add(ref.sessionId)
sessions.push({ ref, status: 'resolved', session: descriptor })
} else {
failedSessionIds.add(ref.sessionId)
sessions.push({ ref, status: 'unresolved' })
unresolved.push({ ref, reason: 'session_not_found' })
}
continue
}
const detail = this.deps.contactsService.getContactDetail(ref.contactKey, { acceptStale: true })
const contact = detail.contact
if (!contact) {
contacts.push({
ref,
status: 'unresolved',
cacheStatus: detail.cache.status,
sessions: [],
unresolvedSessionIds: [],
failedSessionIds: [],
})
unresolved.push({
ref,
reason: detail.cache.status === 'missing' ? 'contact_snapshot_missing' : 'contact_not_found',
})
continue
}
const sourceSessions = contact.sessionScoped
? contact.sourceSessions.filter((source) => source.id === contact.sessionId)
: contact.sourceSessions
const resolvedContactSessions: CrossChatResolvedContact['sessions'] = []
const unresolvedContactSessions: string[] = []
const failedContactSessions: string[] = []
for (const source of sourceSessions) {
candidateSessionIds.add(source.id)
try {
const db = this.deps.adapter.openReadonly(source.id)
if (!db) {
failedContactSessions.push(source.id)
failedSessionIds.add(source.id)
continue
}
const descriptor = getSessionDescriptor(source.id, db)
const member = getMembers(db).find((item) => item.platformId === contact.platformId)
if (!descriptor || !member) {
unresolvedContactSessions.push(source.id)
continue
}
resolvedContactSessions.push({
...descriptor,
memberId: member.id,
memberPlatformId: member.platformId,
memberName: member.name,
})
resolvedSessionIds.add(source.id)
} catch (error) {
failedContactSessions.push(source.id)
failedSessionIds.add(source.id)
appLogger.warn('cross-chat-analysis', `failed to resolve contact in source session: ${source.id}`, error)
}
}
const status =
resolvedContactSessions.length === 0
? 'unresolved'
: unresolvedContactSessions.length > 0 || failedContactSessions.length > 0
? 'partial'
: 'resolved'
contacts.push({
ref,
status,
cacheStatus: detail.cache.status,
sessions: resolvedContactSessions,
unresolvedSessionIds: unresolvedContactSessions,
failedSessionIds: failedContactSessions,
})
if (status === 'unresolved') unresolved.push({ ref, reason: 'member_not_found' })
}
return {
contacts,
sessions,
unresolved,
coverage: {
requestedEntities: refs.length,
resolvedEntities:
contacts.filter((item) => item.status !== 'unresolved').length +
sessions.filter((item) => item.status === 'resolved').length,
candidateSessions: candidateSessionIds.size,
resolvedSessions: resolvedSessionIds.size,
failedSessions: failedSessionIds.size,
},
}
}
async searchMessages(
request: CrossChatSearchRequest,
options: CrossChatOperationOptions = {}
): Promise<CrossChatSearchResult> {
const keywords = request.keywords.map((keyword) => keyword.trim()).filter(Boolean)
if (keywords.length === 0) throw new Error('At least one search keyword is required')
throwIfAborted(options.signal)
const maxSessions = clampInteger(request.maxSessions, DEFAULT_MAX_SESSIONS, 1, MAX_MAX_SESSIONS)
const maxEvidence = clampInteger(request.maxEvidence, DEFAULT_MAX_EVIDENCE, 1, MAX_MAX_EVIDENCE)
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(request.scopes, failedSessionIds)
const selected = candidates.slice(0, maxSessions)
const truncatedReasons = new Set<CrossChatTruncationReason>()
if (candidates.length > selected.length) truncatedReasons.add('session_budget')
const messages: CrossChatMessageSource[] = []
let totalMatches = 0
let scannedSessions = 0
let matchedSessions = 0
let processedSessions = 0
options.onProgress?.({ processedSessions: 0, totalSessions: selected.length })
for (const [index, candidate] of selected.entries()) {
throwIfAborted(options.signal)
if (this.now() - startedAt >= maxWallTimeMs) {
truncatedReasons.add('time_budget')
break
}
options.onProgress?.({
processedSessions: index,
totalSessions: selected.length,
currentSessionId: candidate.descriptor.sessionId,
})
throwIfAborted(options.signal)
try {
const db = this.deps.adapter.openReadonly(candidate.descriptor.sessionId)
if (!db) {
failedSessionIds.add(candidate.descriptor.sessionId)
continue
}
const result = searchMessagesByKeywords(db, keywords, {
startTs: request.startTs,
endTs: request.endTs,
senderIds: candidate.memberIds,
matchMode: request.matchMode,
sort: request.sort,
limit: maxEvidence,
})
scannedSessions++
totalMatches += result.total ?? result.messages.length
if (result.messages.length > 0) matchedSessions++
if ((result.total ?? result.messages.length) > result.messages.length) truncatedReasons.add('evidence_budget')
messages.push(...result.messages.map((message) => toCrossChatMessage(candidate.descriptor, message)))
} catch (error) {
failedSessionIds.add(candidate.descriptor.sessionId)
appLogger.warn('cross-chat-analysis', `failed to search session: ${candidate.descriptor.sessionId}`, error)
} finally {
processedSessions++
}
await yieldToEventLoop()
}
const sortMultiplier = request.sort === 'asc' ? 1 : -1
messages.sort(
(left, right) =>
(left.timestamp - right.timestamp ||
left.sessionId.localeCompare(right.sessionId) ||
left.messageId - right.messageId) * sortMultiplier
)
if (messages.length > maxEvidence) {
messages.length = maxEvidence
truncatedReasons.add('evidence_budget')
}
options.onProgress?.({ processedSessions, totalSessions: selected.length })
return {
messages,
totalMatches,
coverage: {
candidateSessions: candidateSessionCount,
scannedSessions,
matchedSessions,
failedSessions: failedSessionIds.size,
truncated: truncatedReasons.size > 0,
truncatedReasons: [...truncatedReasons],
},
}
}
getMessageContext(request: CrossChatMessageContextRequest): CrossChatMessageContextResult {
const db = this.deps.adapter.ensureReadonly(request.sessionId)
const descriptor = getSessionDescriptor(request.sessionId, db)
if (!descriptor) throw createNotFoundError(`Session not found: ${request.sessionId}`)
const contextSize = clampInteger(request.contextSize, DEFAULT_CONTEXT_SIZE, 0, MAX_CONTEXT_SIZE)
const messages = getCoreMessageContext(db, [request.messageId], contextSize)
if (!messages.some((message) => message.id === request.messageId)) {
throw createNotFoundError(`Message not found: ${request.messageId}`)
}
return {
source: descriptor,
messages: messages.map((message) => toCrossChatMessage(descriptor, message)),
}
}
async getOverview(
request: CrossChatOverviewRequest,
options: CrossChatOperationOptions = {}
): Promise<CrossChatOverviewResult> {
throwIfAborted(options.signal)
const scopes = normalizeScopes(request.scopes)
const maxSessions = clampInteger(request.maxSessions, DEFAULT_MAX_SESSIONS, 1, MAX_MAX_SESSIONS)
const maxWallTimeMs = clampInteger(request.maxWallTimeMs, DEFAULT_MAX_WALL_TIME_MS, 1, MAX_MAX_WALL_TIME_MS)
const selected = scopes.slice(0, maxSessions)
const startedAt = this.now()
const items: CrossChatOverviewItem[] = []
let failedSessions = 0
let processedSessions = 0
const truncatedReasons = new Set<'session_budget' | 'time_budget'>()
if (selected.length < scopes.length) truncatedReasons.add('session_budget')
options.onProgress?.({ processedSessions: 0, totalSessions: selected.length })
for (const [index, scope] of selected.entries()) {
throwIfAborted(options.signal)
if (this.now() - startedAt >= maxWallTimeMs) {
truncatedReasons.add('time_budget')
break
}
options.onProgress?.({
processedSessions: index,
totalSessions: selected.length,
currentSessionId: scope.sessionId,
})
throwIfAborted(options.signal)
try {
const db = this.deps.adapter.openReadonly(scope.sessionId)
if (!db) {
failedSessions++
continue
}
const descriptor = getSessionDescriptor(scope.sessionId, db)
if (!descriptor) {
failedSessions++
continue
}
items.push(buildOverviewItem(descriptor, db, scope))
} catch (error) {
failedSessions++
appLogger.warn('cross-chat-analysis', `failed to build session overview: ${scope.sessionId}`, error)
} finally {
processedSessions++
}
await yieldToEventLoop()
}
options.onProgress?.({ processedSessions, totalSessions: selected.length })
return {
items,
coverage: {
candidateSessions: scopes.length,
analyzedSessions: items.length,
failedSessions,
truncated: truncatedReasons.size > 0,
truncatedReasons: [...truncatedReasons],
},
}
}
private resolveSearchCandidates(
scopes: CrossChatSearchScope[] | undefined,
failedSessionIds: Set<string>
): { candidates: SearchCandidate[]; candidateSessionCount: number } {
const normalizedScopes: CrossChatSearchScope[] = scopes
? normalizeScopes(scopes)
: this.deps.adapter.listSessionIds().map((sessionId) => ({ sessionId }))
const candidates: SearchCandidate[] = []
for (const scope of normalizedScopes) {
const descriptor = this.tryGetSessionDescriptor(scope.sessionId)
if (!descriptor) {
failedSessionIds.add(scope.sessionId)
continue
}
candidates.push({ descriptor, memberIds: scope.memberIds })
}
return {
candidates: candidates.sort(
(left, right) =>
(right.descriptor.lastMessageTs ?? Number.NEGATIVE_INFINITY) -
(left.descriptor.lastMessageTs ?? Number.NEGATIVE_INFINITY)
),
candidateSessionCount: normalizedScopes.length,
}
}
private tryGetSessionDescriptor(sessionId: string): CrossChatSessionDescriptor | null {
try {
const db = this.deps.adapter.openReadonly(sessionId)
return db ? getSessionDescriptor(sessionId, db) : null
} catch (error) {
appLogger.warn('cross-chat-analysis', `failed to inspect session metadata: ${sessionId}`, error)
return null
}
}
private now(): number {
return this.deps.now?.() ?? Date.now()
}
}
interface SearchCandidate {
descriptor: CrossChatSessionDescriptor
memberIds?: number[]
}
function getSessionDescriptor(sessionId: string, db: DatabaseAdapter): CrossChatSessionDescriptor | null {
const meta = getSessionMeta(db)
if (!meta || (meta.type !== ChatType.PRIVATE && meta.type !== ChatType.GROUP)) return null
const latest = getRecentMessages(db, { limit: 1 })[0]
return {
sessionId,
sessionName: meta.name,
sessionType: meta.type,
platform: meta.platform,
lastMessageTs: latest?.timestamp ?? null,
}
}
function toCrossChatMessage(
descriptor: CrossChatSessionDescriptor,
message: {
id: number
senderId: number
senderName: string
senderPlatformId: string
content: string
timestamp: number
type: number
}
): CrossChatMessageSource {
return {
...descriptor,
messageId: message.id,
senderId: message.senderId,
senderName: message.senderName,
senderPlatformId: message.senderPlatformId,
content: message.content,
timestamp: message.timestamp,
messageType: message.type,
}
}
function normalizeScopes(scopes: CrossChatSearchScope[]): CrossChatSearchScope[] {
const bySession = new Map<string, CrossChatSearchScope>()
for (const scope of scopes) {
const sessionId = scope.sessionId.trim()
if (!sessionId) continue
const existing = bySession.get(sessionId)
if (!existing) {
bySession.set(sessionId, {
sessionId,
memberIds: scope.memberIds ? [...new Set(scope.memberIds)] : undefined,
label: scope.label,
})
continue
}
if (!existing.memberIds || !scope.memberIds) {
existing.memberIds = undefined
} else {
existing.memberIds = [...new Set([...existing.memberIds, ...scope.memberIds])]
}
existing.label ??= scope.label
}
return [...bySession.values()]
}
function buildOverviewItem(
descriptor: CrossChatSessionDescriptor,
db: DatabaseAdapter,
scope: CrossChatSearchScope
): CrossChatOverviewItem {
if (!scope.memberIds) {
const overview = getSessionOverview(db)
return {
...descriptor,
label: scope.label ?? descriptor.sessionName,
totalMessages: overview.totalMessages,
firstMessageTs: overview.firstMessageTs,
lastMessageTs: overview.lastMessageTs,
}
}
const members = getMembers(db)
const memberNames = members.filter((member) => scope.memberIds?.includes(member.id)).map((member) => member.name)
const latest = searchMessagesByKeywords(db, [], { senderIds: scope.memberIds, sort: 'desc', limit: 1 })
const earliest = searchMessagesByKeywords(db, [], { senderIds: scope.memberIds, sort: 'asc', limit: 1 })
return {
...descriptor,
label: scope.label ?? descriptor.sessionName,
memberIds: scope.memberIds,
memberNames,
totalMessages: latest.total ?? 0,
firstMessageTs: earliest.messages[0]?.timestamp ?? null,
lastMessageTs: latest.messages[0]?.timestamp ?? null,
}
}
function clampInteger(value: number | undefined, fallback: number, min: number, max: number): number {
if (!Number.isFinite(value)) return fallback
return Math.min(max, Math.max(min, Math.floor(value as number)))
}
function throwIfAborted(signal?: AbortSignal): void {
if (!signal?.aborted) return
const error = new Error('Cross-chat analysis was interrupted')
error.name = 'AbortError'
throw error
}
function yieldToEventLoop(): Promise<void> {
return new Promise((resolve) => setImmediate(resolve))
}
function createNotFoundError(message: string): Error {
return Object.assign(new Error(message), { statusCode: 404 })
}
@@ -0,0 +1,140 @@
import type { ChatPlatform, ChatType, ContactsCacheStatus } from '@openchatlab/shared-types'
import type { AIEntityRef } from '../../ai/chats'
export interface CrossChatSessionDescriptor {
sessionId: string
sessionName: string
sessionType: ChatType
platform: ChatPlatform
lastMessageTs: number | null
}
export interface CrossChatResolvedContactSession extends CrossChatSessionDescriptor {
memberId: number
memberPlatformId: string
memberName: string
}
export interface CrossChatResolvedContact {
ref: Extract<AIEntityRef, { type: 'contact' }>
status: 'resolved' | 'partial' | 'unresolved'
cacheStatus: ContactsCacheStatus
sessions: CrossChatResolvedContactSession[]
unresolvedSessionIds: string[]
failedSessionIds: string[]
}
export interface CrossChatResolvedSession {
ref: Extract<AIEntityRef, { type: 'session' }>
status: 'resolved' | 'unresolved'
session?: CrossChatSessionDescriptor
}
export interface CrossChatUnresolvedEntity {
ref: AIEntityRef
reason: 'contact_snapshot_missing' | 'contact_not_found' | 'session_not_found' | 'member_not_found'
}
export interface CrossChatEntityResolution {
contacts: CrossChatResolvedContact[]
sessions: CrossChatResolvedSession[]
unresolved: CrossChatUnresolvedEntity[]
coverage: {
requestedEntities: number
resolvedEntities: number
candidateSessions: number
resolvedSessions: number
failedSessions: number
}
}
export interface CrossChatSearchScope {
sessionId: string
memberIds?: number[]
label?: string
}
export interface CrossChatSearchRequest {
keywords: string[]
scopes?: CrossChatSearchScope[]
startTs?: number
endTs?: number
matchMode?: 'any' | 'all'
sort?: 'asc' | 'desc'
maxSessions?: number
maxEvidence?: number
maxWallTimeMs?: number
}
export interface CrossChatSearchProgress {
processedSessions: number
totalSessions: number
currentSessionId?: string
}
export interface CrossChatOperationOptions {
signal?: AbortSignal
onProgress?: (progress: CrossChatSearchProgress) => void
}
export interface CrossChatMessageSource extends CrossChatSessionDescriptor {
messageId: number
senderId: number
senderName: string
senderPlatformId: string
content: string
timestamp: number
messageType: number
}
export type CrossChatTruncationReason = 'session_budget' | 'evidence_budget' | 'time_budget'
export interface CrossChatSearchResult {
messages: CrossChatMessageSource[]
totalMatches: number
coverage: {
candidateSessions: number
scannedSessions: number
matchedSessions: number
failedSessions: number
truncated: boolean
truncatedReasons: CrossChatTruncationReason[]
}
}
export interface CrossChatMessageContextRequest {
sessionId: string
messageId: number
contextSize?: number
}
export interface CrossChatMessageContextResult {
source: CrossChatSessionDescriptor
messages: CrossChatMessageSource[]
}
export interface CrossChatOverviewRequest {
scopes: CrossChatSearchScope[]
maxSessions?: number
maxWallTimeMs?: number
}
export interface CrossChatOverviewItem extends CrossChatSessionDescriptor {
label: string
memberIds?: number[]
memberNames?: string[]
totalMessages: number
firstMessageTs: number | null
lastMessageTs: number | null
}
export interface CrossChatOverviewResult {
items: CrossChatOverviewItem[]
coverage: {
candidateSessions: number
analyzedSessions: number
failedSessions: number
truncated: boolean
truncatedReasons: Array<'session_budget' | 'time_budget'>
}
}
@@ -53,6 +53,31 @@ export { exportMarkdown } from './export-service'
export { CONTACTS_ALGORITHM_VERSION, createContactsService } from './contacts'
export type { ContactsComputeRunner, ContactsService, ContactsServiceDeps, ContactsServiceOptions } from './contacts'
// Cross-chat AI analysis service
export { createCrossChatAnalysisService } from './cross-chat-analysis'
export type {
CrossChatAnalysisService,
CrossChatAnalysisServiceDeps,
CrossChatEntityResolution,
CrossChatMessageContextRequest,
CrossChatMessageContextResult,
CrossChatMessageSource,
CrossChatOperationOptions,
CrossChatOverviewItem,
CrossChatOverviewRequest,
CrossChatOverviewResult,
CrossChatResolvedContact,
CrossChatResolvedContactSession,
CrossChatResolvedSession,
CrossChatSearchProgress,
CrossChatSearchRequest,
CrossChatSearchResult,
CrossChatSearchScope,
CrossChatSessionDescriptor,
CrossChatTruncationReason,
CrossChatUnresolvedEntity,
} from './cross-chat-analysis'
// People relationships service
export { PEOPLE_RELATIONSHIPS_ALGORITHM_VERSION, createPeopleRelationshipsService } from './people/relationships'
export type {