diff --git a/packages/core/src/query/__tests__/message-search-extensions.test.ts b/packages/core/src/query/__tests__/message-search-extensions.test.ts index 60a9c70b..8e5ee455 100644 --- a/packages/core/src/query/__tests__/message-search-extensions.test.ts +++ b/packages/core/src/query/__tests__/message-search-extensions.test.ts @@ -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)', () => { diff --git a/packages/core/src/query/message-queries.ts b/packages/core/src/query/message-queries.ts index 50789762..295f6c5a 100644 --- a/packages/core/src/query/message-queries.ts +++ b/packages/core/src/query/message-queries.ts @@ -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, diff --git a/packages/core/src/query/message-sql.ts b/packages/core/src/query/message-sql.ts index aee8800f..78fb8251 100644 --- a/packages/core/src/query/message-sql.ts +++ b/packages/core/src/query/message-sql.ts @@ -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 = ?') diff --git a/packages/node-runtime/src/index.ts b/packages/node-runtime/src/index.ts index 0a48279d..0134391f 100644 --- a/packages/node-runtime/src/index.ts +++ b/packages/node-runtime/src/index.ts @@ -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, diff --git a/packages/node-runtime/src/services/cross-chat-analysis/index.ts b/packages/node-runtime/src/services/cross-chat-analysis/index.ts new file mode 100644 index 00000000..9cee27fb --- /dev/null +++ b/packages/node-runtime/src/services/cross-chat-analysis/index.ts @@ -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' diff --git a/packages/node-runtime/src/services/cross-chat-analysis/service.test.ts b/packages/node-runtime/src/services/cross-chat-analysis/service.test.ts new file mode 100644 index 00000000..9bbd7d85 --- /dev/null +++ b/packages/node-runtime/src/services/cross-chat-analysis/service.test.ts @@ -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() + 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 & Pick +): 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 +} { + 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([ + [ + '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() + } +}) diff --git a/packages/node-runtime/src/services/cross-chat-analysis/service.ts b/packages/node-runtime/src/services/cross-chat-analysis/service.ts new file mode 100644 index 00000000..73c49d08 --- /dev/null +++ b/packages/node-runtime/src/services/cross-chat-analysis/service.ts @@ -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 + now?: () => number +} + +export interface CrossChatAnalysisService { + resolveEntities(refs: AIEntityRef[]): CrossChatEntityResolution + searchMessages(request: CrossChatSearchRequest, options?: CrossChatOperationOptions): Promise + getMessageContext(request: CrossChatMessageContextRequest): CrossChatMessageContextResult + getOverview(request: CrossChatOverviewRequest, options?: CrossChatOperationOptions): Promise +} + +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() + const resolvedSessionIds = new Set() + const failedSessionIds = new Set() + + 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 { + 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() + const { candidates, candidateSessionCount } = this.resolveSearchCandidates(request.scopes, failedSessionIds) + const selected = candidates.slice(0, maxSessions) + const truncatedReasons = new Set() + 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 { + 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 + ): { 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() + 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 { + return new Promise((resolve) => setImmediate(resolve)) +} + +function createNotFoundError(message: string): Error { + return Object.assign(new Error(message), { statusCode: 404 }) +} diff --git a/packages/node-runtime/src/services/cross-chat-analysis/types.ts b/packages/node-runtime/src/services/cross-chat-analysis/types.ts new file mode 100644 index 00000000..8b2e3fad --- /dev/null +++ b/packages/node-runtime/src/services/cross-chat-analysis/types.ts @@ -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 + status: 'resolved' | 'partial' | 'unresolved' + cacheStatus: ContactsCacheStatus + sessions: CrossChatResolvedContactSession[] + unresolvedSessionIds: string[] + failedSessionIds: string[] +} + +export interface CrossChatResolvedSession { + ref: Extract + 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'> + } +} diff --git a/packages/node-runtime/src/services/index.ts b/packages/node-runtime/src/services/index.ts index 7ccbaad0..298980b1 100644 --- a/packages/node-runtime/src/services/index.ts +++ b/packages/node-runtime/src/services/index.ts @@ -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 {