From d1a6c711fe6ef0735bf9943622f89ad5ada8420d Mon Sep 17 00:00:00 2001 From: digua Date: Fri, 21 Aug 2026 23:42:35 +0800 Subject: [PATCH] feat(ai): add isolated cross-chat agent --- apps/cli/src/ai/agent-stream-runner.test.ts | 49 +- apps/cli/src/ai/agent-stream-runner.ts | 65 +++ apps/cli/src/ai/cross-chat-tool-adapter.ts | 48 ++ apps/cli/src/http/index.ts | 17 + apps/cli/src/http/routes/web/index.ts | 3 + apps/desktop/main/ai/agent-stream-runner.ts | 63 ++- .../main/ai/cross-chat-tool-adapter.ts | 48 ++ apps/desktop/main/internal-api/server.ts | 18 +- packages/http-routes/src/context/ai.ts | 5 +- .../src/ai/__tests__/cross-chat-agent.test.ts | 117 +++++ packages/node-runtime/src/ai/chats.ts | 15 +- .../compression/__tests__/compressor.test.ts | 30 ++ .../src/ai/compression/adapter-factory.ts | 5 +- .../src/ai/compression/compressor.ts | 18 +- .../node-runtime/src/ai/compression/types.ts | 2 +- .../node-runtime/src/ai/cross-chat-agent.ts | 242 +++++++++ packages/node-runtime/src/ai/index.ts | 2 + packages/node-runtime/src/index.ts | 3 + .../src/services/cross-chat-analysis/index.ts | 1 + .../cross-chat-analysis/preprocess.ts | 37 ++ .../cross-chat-analysis/service.test.ts | 337 +++++++++++-- .../services/cross-chat-analysis/service.ts | 189 +++++-- .../src/services/cross-chat-analysis/types.ts | 160 +----- packages/node-runtime/src/services/index.ts | 2 +- packages/shared-types/index.ts | 204 ++++++++ packages/tools/src/agent-adapter.ts | 8 +- .../src/definitions/cross-chat-tools.test.ts | 472 ++++++++++++++++++ .../tools/src/definitions/cross-chat-tools.ts | 452 +++++++++++++++++ packages/tools/src/index.ts | 9 + packages/tools/src/registry.ts | 14 + packages/tools/src/types.ts | 45 +- src/services/ai-stream/types.ts | 6 +- 32 files changed, 2443 insertions(+), 243 deletions(-) create mode 100644 apps/cli/src/ai/cross-chat-tool-adapter.ts create mode 100644 apps/desktop/main/ai/cross-chat-tool-adapter.ts create mode 100644 packages/node-runtime/src/ai/__tests__/cross-chat-agent.test.ts create mode 100644 packages/node-runtime/src/ai/cross-chat-agent.ts create mode 100644 packages/node-runtime/src/services/cross-chat-analysis/preprocess.ts create mode 100644 packages/tools/src/definitions/cross-chat-tools.test.ts create mode 100644 packages/tools/src/definitions/cross-chat-tools.ts diff --git a/apps/cli/src/ai/agent-stream-runner.test.ts b/apps/cli/src/ai/agent-stream-runner.test.ts index f9d974c7..1aadf31c 100644 --- a/apps/cli/src/ai/agent-stream-runner.test.ts +++ b/apps/cli/src/ai/agent-stream-runner.test.ts @@ -1,7 +1,8 @@ import { describe, it } from 'node:test' import assert from 'node:assert/strict' +import type { AIChatManager, AgentStreamChunk, DatabaseManager, LLMConfigStore } from '@openchatlab/node-runtime' import { getChartCapabilityAllowedBuiltinTools } from '@openchatlab/node-runtime' -import { getAllowedToolSet, getAvailableToolDefs } from './agent-stream-runner' +import { createCliRunAgentStream, getAllowedToolSet, getAvailableToolDefs } from './agent-stream-runner' describe('CLI chart capability tool filtering', () => { it('does not expose uncategorized raw SQL in chart-only turns', () => { @@ -64,3 +65,49 @@ describe('CLI chart capability tool filtering', () => { assert.ok(!toolNames.includes('execute_sql')) }) }) + +describe('CLI agent conversation target validation', () => { + it('rejects a global conversation before the default session tool path can run', async () => { + let databaseOpenCalls = 0 + const dbManager = { + pathProvider: { getAiDataDir: () => '/tmp/chatlab-agent-runner-test' }, + open() { + databaseOpenCalls++ + throw new Error('session database must not be opened') + }, + } as unknown as DatabaseManager + const aiChatManager = { + getAIChat: () => ({ + id: 'global-chat-1', + sessionId: '', + kind: 'global', + title: null, + assistantId: 'default', + createdAt: 1, + updatedAt: 1, + }), + } as unknown as AIChatManager + const llmConfigStore = { + getDefaultAssistantConfig: () => null, + } as unknown as LLMConfigStore + const events: AgentStreamChunk[] = [] + const runAgentStream = createCliRunAgentStream(dbManager, aiChatManager, { llmConfigStore }) + + await runAgentStream( + { + userMessage: '分析一下', + aiChatId: 'global-chat-1', + sessionId: 'private-session-1', + }, + (event) => events.push(event), + new AbortController().signal + ) + + assert.equal(databaseOpenCalls, 0) + assert.deepEqual( + events.map((event) => event.type), + ['error', 'done'] + ) + assert.match(String(events[0]?.error && (events[0].error as { message?: string }).message), /does not match/) + }) +}) diff --git a/apps/cli/src/ai/agent-stream-runner.ts b/apps/cli/src/ai/agent-stream-runner.ts index c2b1276e..7204a34a 100644 --- a/apps/cli/src/ai/agent-stream-runner.ts +++ b/apps/cli/src/ai/agent-stream-runner.ts @@ -9,6 +9,8 @@ import type { AgentStreamChunk, LLMConfigStore, SemanticIndexRuntime, + CrossChatAnalysisService, + SessionRuntimeAdapter, } from '@openchatlab/node-runtime' import { CHART_CAPABILITY_CORE_TOOLS, @@ -23,6 +25,7 @@ import { getChartCapabilitySkill, getSkillConfigWithBuiltinChart, resolveChartRuntimeForRequest, + runCrossChatAgent, } from '@openchatlab/node-runtime' import { getChatOverview, normalizeBuiltinToolNames } from '@openchatlab/core' import type { ChartAutoMode } from '@openchatlab/shared-types' @@ -33,6 +36,8 @@ import { buildSemanticSearchGuidance } from '@openchatlab/node-runtime' import { adaptToolsForAgent } from './tool-adapter' import { loadAssistantConfig } from './assistant-loader' import { runServerAgent } from './agent' +import { createCliCrossChatTools } from './cross-chat-tool-adapter' +import { getServerAiLogger } from './logger' function getAiDir(dbManager: DatabaseManager): string { const pathProvider = (dbManager as any)['pathProvider'] @@ -76,6 +81,8 @@ export function createCliRunAgentStream( options: { llmConfigStore?: LLMConfigStore semanticIndexService?: SemanticIndexRuntime + crossChatAnalysisService?: CrossChatAnalysisService + sessionAdapter?: SessionRuntimeAdapter } = {} ): (params: AgentStreamRequest, onEvent: (chunk: AgentStreamChunk) => void, abortSignal: AbortSignal) => Promise { const aiDataDir = getAiDir(dbManager) @@ -87,7 +94,9 @@ export function createCliRunAgentStream( userMessage, aiChatId, historyLeafMessageId, + chatKind = 'session', sessionId, + entityRefs, chatType, locale, assistantId, @@ -99,6 +108,20 @@ export function createCliRunAgentStream( thinkingLevel, } = params + const conversation = aiChatManager.getAIChat(aiChatId) + if ( + !conversation || + conversation.kind !== chatKind || + (chatKind === 'session' && conversation.sessionId !== sessionId) + ) { + onEvent({ + type: 'error', + error: { name: 'ValidationError', message: 'AI conversation does not match the requested target' }, + }) + onEvent({ type: 'done', isFinished: true }) + return + } + let assistantSystemPrompt: string | undefined let assistantAllowedTools: string[] | undefined if (assistantId) { @@ -114,6 +137,48 @@ export function createCliRunAgentStream( const contextWindow = llmConfig ? (buildPiModel(llmConfig).contextWindow ?? 128000) : 128000 const maxToolResultTokens = Math.floor(contextWindow * (maxToolResultPercent / 100)) + if (chatKind === 'global') { + if (!llmConfig) { + onEvent({ type: 'error', error: { name: 'ConfigError', message: 'LLM service not configured' } }) + onEvent({ type: 'done', isFinished: true }) + return + } + if (!options.crossChatAnalysisService || !options.sessionAdapter) { + onEvent({ type: 'error', error: { name: 'RuntimeError', message: 'Cross-chat analysis is unavailable' } }) + onEvent({ type: 'done', isFinished: true }) + return + } + const tools = createCliCrossChatTools({ + analysisService: options.crossChatAnalysisService, + sessionAdapter: options.sessionAdapter, + locale, + preprocessConfig: params.preprocessConfig, + maxToolResultTokens, + }) + await runCrossChatAgent({ + userMessage, + entityRefs, + aiChatId, + historyLeafMessageId, + locale, + piModel: buildPiModel(llmConfig), + apiKey: llmConfig.apiKey, + tools, + aiChatManager, + onEvent, + abortSignal, + thinkingLevel: thinkingLevel as import('@openchatlab/core').ThinkingLevel | undefined, + logger: getServerAiLogger() ?? undefined, + }) + return + } + + if (!sessionId) { + onEvent({ type: 'error', error: { name: 'ValidationError', message: 'Session ID is required' } }) + onEvent({ type: 'done', isFinished: true }) + return + } + const db = (dbManager as any).open?.(sessionId) const resolvedChartAutoMode: ChartAutoMode = chartAutoMode ?? 'suggest' const chartRuntime = resolveChartRuntimeForRequest({ diff --git a/apps/cli/src/ai/cross-chat-tool-adapter.ts b/apps/cli/src/ai/cross-chat-tool-adapter.ts new file mode 100644 index 00000000..7ba479f2 --- /dev/null +++ b/apps/cli/src/ai/cross-chat-tool-adapter.ts @@ -0,0 +1,48 @@ +import type { + AgentTool, + CrossChatAnalysisService, + PreprocessConfig, + SessionRuntimeAdapter, +} from '@openchatlab/node-runtime' +import { preprocessCrossChatMessages } from '@openchatlab/node-runtime' +import { + CROSS_CHAT_AGENT_TOOL_REGISTRY, + executeToolForAgent, + toAgentToolParameters, + type CrossChatToolExecutionContext, +} from '@openchatlab/tools' + +export function createCliCrossChatTools(options: { + analysisService: CrossChatAnalysisService + sessionAdapter: SessionRuntimeAdapter + locale?: string + preprocessConfig?: Record + maxToolResultTokens: number +}): AgentTool[] { + const context: Omit = { + locale: options.locale, + analysisService: options.analysisService, + maxToolResultTokens: options.maxToolResultTokens, + preprocessMessagesBySession: (sessionId, messages) => + preprocessCrossChatMessages( + options.sessionAdapter, + sessionId, + messages, + options.preprocessConfig as PreprocessConfig | undefined + ), + } + + return CROSS_CHAT_AGENT_TOOL_REGISTRY.map((tool) => ({ + name: tool.name, + label: tool.name, + description: tool.description, + parameters: toAgentToolParameters(tool.inputSchema) as any, + executionMode: tool.executionMode, + execute: async (_toolCallId: string, params: unknown, signal, onUpdate) => + executeToolForAgent(tool, params, { + ...context, + abortSignal: signal, + reportProgress: (progress) => onUpdate?.({ content: [], details: { progress } }), + }), + })) +} diff --git a/apps/cli/src/http/index.ts b/apps/cli/src/http/index.ts index d1ed8e53..45bc0d68 100644 --- a/apps/cli/src/http/index.ts +++ b/apps/cli/src/http/index.ts @@ -22,6 +22,9 @@ import { initAppLogger, appLogger, logNativeParserStatus, + createContactsService, + createCrossChatAnalysisService, + createDatabaseManagerAdapter, } from '@openchatlab/node-runtime' import type { SemanticIndexRuntime } from '@openchatlab/node-runtime' import { createServer } from './server' @@ -136,6 +139,17 @@ export async function startHttpServer(options?: HttpServerOptions): Promise<{ await migrationRunner.run() const nativeBinding = resolveNativeBinding() dbManager = new DatabaseManager(pathProvider, { nativeBinding, runtime }) + const sessionAdapter = createDatabaseManagerAdapter(dbManager) + const contactsService = createContactsService({ + adapter: sessionAdapter, + pathProvider, + runtimeIdentity: runtime, + nativeBinding, + }) + const crossChatAnalysisService = createCrossChatAnalysisService({ + adapter: sessionAdapter, + contactsService, + }) const aiDataDir = pathProvider.getAiDataDir() aiChatManager = new AIChatManager(aiDataDir, { nativeBinding }) @@ -185,6 +199,7 @@ export async function startHttpServer(options?: HttpServerOptions): Promise<{ pathProvider, nativeBinding, runtimeIdentity: runtime, + contactsService, semanticIndexService, aiContext: { aiDataDir, @@ -197,6 +212,8 @@ export async function startHttpServer(options?: HttpServerOptions): Promise<{ runAgentStream: createCliRunAgentStream(dbManager, aiChatManager, { llmConfigStore, semanticIndexService, + crossChatAnalysisService, + sessionAdapter, }), }, }) diff --git a/apps/cli/src/http/routes/web/index.ts b/apps/cli/src/http/routes/web/index.ts index 58be6790..7e8efa2b 100644 --- a/apps/cli/src/http/routes/web/index.ts +++ b/apps/cli/src/http/routes/web/index.ts @@ -29,6 +29,7 @@ import type { LLMConfigStore, CustomProviderStore, CustomModelStore, + ContactsService, PendingDataDirMigration, RuntimeIdentity, } from '@openchatlab/node-runtime' @@ -71,6 +72,7 @@ export function registerWebRoutes( pathProvider?: PathProvider nativeBinding?: string runtimeIdentity?: RuntimeIdentity + contactsService?: ContactsService aiContext?: AiContextOptions /** 由 server 入口注入的共享语义索引运行时；传入时由调用方管理生命周期 */ semanticIndexService?: SemanticIndexRuntime @@ -132,6 +134,7 @@ export function registerWebRoutes( getVersion, nativeBinding: options?.nativeBinding, semanticIndexService, + contactsService: options?.contactsService, analyticsService, getCurrentAiLogPath: () => getServerAiLogger()?.getExistingLogPath() ?? null, openDirectory: openDirectoryPath, diff --git a/apps/desktop/main/ai/agent-stream-runner.ts b/apps/desktop/main/ai/agent-stream-runner.ts index 10059b5a..92ceb88b 100644 --- a/apps/desktop/main/ai/agent-stream-runner.ts +++ b/apps/desktop/main/ai/agent-stream-runner.ts @@ -11,6 +11,8 @@ import type { AIServiceConfig, LLMConfigStore, SemanticIndexRuntime, + CrossChatAnalysisService, + SessionRuntimeAdapter, } from '@openchatlab/node-runtime' import type { AgentStreamRequest } from '@openchatlab/http-routes' import { getDefaultGeneralAssistantId, type ChartAutoMode } from '@openchatlab/shared-types' @@ -27,6 +29,7 @@ import { initTokenizer, resolveChartRuntimeForRequest, buildSemanticSearchGuidance, + runCrossChatAgent, } from '@openchatlab/node-runtime' import type { CompressionLlmAdapter, AgentRuntimeStatus } from '@openchatlab/node-runtime' import { Agent, type AgentStreamChunk, type SkillContext } from './agent' @@ -41,6 +44,7 @@ import { getManager as getAIChatManager } from './chats' import { t } from '../i18n' import * as workerManager from '../worker/workerManager' import { getProviderInfo, type LLMProvider } from './llm' +import { createElectronCrossChatTools } from './cross-chat-tool-adapter' const DEFAULT_CONTEXT_WINDOW = 128000 @@ -68,7 +72,11 @@ const compressionLogger = { export function createElectronRunAgentStream( llmConfigStore: LLMConfigStore, - semanticIndexService?: SemanticIndexRuntime + semanticIndexService?: SemanticIndexRuntime, + crossChatOptions?: { + analysisService: CrossChatAnalysisService + sessionAdapter: SessionRuntimeAdapter + } ): ( params: AgentStreamRequest, onEvent: (chunk: SharedAgentStreamChunk) => void, @@ -79,7 +87,9 @@ export function createElectronRunAgentStream( userMessage, aiChatId, historyLeafMessageId, + chatKind = 'session', sessionId, + entityRefs, chatType, locale, assistantId, @@ -94,6 +104,19 @@ export function createElectronRunAgentStream( // 确保 tokenizer rank 表已加载（compression + agent 路径均依赖） await initTokenizer() + const conversation = getAIChatManager().getAIChat(aiChatId) + if ( + !conversation || + conversation.kind !== chatKind || + (chatKind === 'session' && conversation.sessionId !== sessionId) + ) { + onEvent({ + type: 'error', + error: { name: 'ValidationError', message: 'AI conversation does not match the requested target' }, + }) + return + } + const requestId = `internal_${Date.now()}` aiLogger.info('AgentStream', `Agent stream request: ${requestId}`, { @@ -113,6 +136,44 @@ export function createElectronRunAgentStream( } const piModel = buildPiModel(activeAIConfig) + if (chatKind === 'global') { + if (!crossChatOptions) { + onEvent({ type: 'error', error: { name: 'RuntimeError', message: 'Cross-chat analysis is unavailable' } }) + return + } + const modelDef = findModelDefinition(activeAIConfig.provider, activeAIConfig.model || '') + const resolvedContextWindow = modelDef?.contextWindow || DEFAULT_CONTEXT_WINDOW + const maxToolResultPercent = DEFAULT_CONTEXT_COMPRESSION_CONFIG.maxToolResultPercent ?? 50 + const tools = createElectronCrossChatTools({ + analysisService: crossChatOptions.analysisService, + sessionAdapter: crossChatOptions.sessionAdapter, + locale, + preprocessConfig: params.preprocessConfig, + maxToolResultTokens: Math.floor(resolvedContextWindow * (maxToolResultPercent / 100)), + }) + await runCrossChatAgent({ + userMessage, + entityRefs, + aiChatId, + historyLeafMessageId, + locale, + piModel, + apiKey: activeAIConfig.apiKey, + tools, + aiChatManager: getAIChatManager(), + onEvent, + abortSignal, + thinkingLevel: thinkingLevel as import('@openchatlab/core').ThinkingLevel | undefined, + logger: aiLogger, + }) + return + } + + if (!sessionId) { + onEvent({ type: 'error', error: { name: 'ValidationError', message: 'Session ID is required' } }) + return + } + if (aiChatId && historyLeafMessageId === undefined) { try { const tempAssistantConfig = assistantId diff --git a/apps/desktop/main/ai/cross-chat-tool-adapter.ts b/apps/desktop/main/ai/cross-chat-tool-adapter.ts new file mode 100644 index 00000000..5090987b --- /dev/null +++ b/apps/desktop/main/ai/cross-chat-tool-adapter.ts @@ -0,0 +1,48 @@ +import type { + AgentTool, + CrossChatAnalysisService, + PreprocessConfig, + SessionRuntimeAdapter, +} from '@openchatlab/node-runtime' +import { preprocessCrossChatMessages } from '@openchatlab/node-runtime' +import { + CROSS_CHAT_AGENT_TOOL_REGISTRY, + executeToolForAgent, + toAgentToolParameters, + type CrossChatToolExecutionContext, +} from '@openchatlab/tools' + +export function createElectronCrossChatTools(options: { + analysisService: CrossChatAnalysisService + sessionAdapter: SessionRuntimeAdapter + locale?: string + preprocessConfig?: Record + maxToolResultTokens: number +}): AgentTool[] { + const context: Omit = { + locale: options.locale, + analysisService: options.analysisService, + maxToolResultTokens: options.maxToolResultTokens, + preprocessMessagesBySession: (sessionId, messages) => + preprocessCrossChatMessages( + options.sessionAdapter, + sessionId, + messages, + options.preprocessConfig as PreprocessConfig | undefined + ), + } + + return CROSS_CHAT_AGENT_TOOL_REGISTRY.map((tool) => ({ + name: tool.name, + label: tool.name, + description: tool.description, + parameters: toAgentToolParameters(tool.inputSchema) as any, + executionMode: tool.executionMode, + execute: async (_toolCallId: string, params: unknown, signal, onUpdate) => + executeToolForAgent(tool, params, { + ...context, + abortSignal: signal, + reportProgress: (progress) => onUpdate?.({ content: [], details: { progress } }), + }), + })) +} diff --git a/apps/desktop/main/internal-api/server.ts b/apps/desktop/main/internal-api/server.ts index 7d87ca3a..443916c1 100644 --- a/apps/desktop/main/internal-api/server.ts +++ b/apps/desktop/main/internal-api/server.ts @@ -25,6 +25,8 @@ import { streamingImport, createSemanticIndexWorkerRuntimeClient, appLogger, + createContactsService, + createCrossChatAnalysisService, } from '@openchatlab/node-runtime' import type { StreamImportDeps, SemanticIndexRuntime } from '@openchatlab/node-runtime' import { getLoadablePath as getSqliteVecLoadablePath } from 'sqlite-vec' @@ -91,6 +93,16 @@ export async function startInternalServer( newDbManager = new DatabaseManager(pathProvider, { runtime, nativeBinding }) const sessionAdapter = createDatabaseManagerAdapter(newDbManager) + const contactsService = createContactsService({ + adapter: sessionAdapter, + pathProvider, + runtimeIdentity: runtime, + nativeBinding, + }) + const crossChatAnalysisService = createCrossChatAnalysisService({ + adapter: sessionAdapter, + contactsService, + }) const aiDataDir = pathProvider.getAiDataDir() const llmRuntimeStores = getDesktopLlmRuntimeStores(aiDataDir) @@ -179,6 +191,7 @@ export async function startInternalServer( stopTimer, }, semanticIndexService: newSemanticIndexService ?? undefined, + contactsService, openDirectory: (dirPath) => shell.openPath(dirPath).then(() => {}), showInFolder: (filePath) => { shell.showItemInFolder(filePath) @@ -193,7 +206,10 @@ export async function startInternalServer( deletePendingDataDirCleanup: (cleanupId) => { return deletePendingDataDirCleanup(pathProvider.getSystemDir(), pathProvider.getUserDataDir(), cleanupId) }, - runAgentStream: createElectronRunAgentStream(llmConfigStore, newSemanticIndexService ?? undefined), + runAgentStream: createElectronRunAgentStream(llmConfigStore, newSemanticIndexService ?? undefined, { + analysisService: crossChatAnalysisService, + sessionAdapter, + }), executeAiTool: createExecuteElectronAiTool(newSemanticIndexService ?? undefined), } diff --git a/packages/http-routes/src/context/ai.ts b/packages/http-routes/src/context/ai.ts index 70e0c18e..8ea0619b 100644 --- a/packages/http-routes/src/context/ai.ts +++ b/packages/http-routes/src/context/ai.ts @@ -10,6 +10,7 @@ import type { SkillManagerCore, } from '@openchatlab/node-runtime' import type { ChartAutoMode } from '@openchatlab/shared-types' +import type { AIEntityRef } from '@openchatlab/shared-types' export interface AiToolExecuteRequest { testId: string @@ -32,7 +33,9 @@ export interface AgentStreamRequest { userMessage: string aiChatId: string historyLeafMessageId?: string | null - sessionId: string + chatKind?: 'session' | 'global' + sessionId?: string + entityRefs?: AIEntityRef[] chatType?: 'group' | 'private' locale?: string assistantId?: string diff --git a/packages/node-runtime/src/ai/__tests__/cross-chat-agent.test.ts b/packages/node-runtime/src/ai/__tests__/cross-chat-agent.test.ts new file mode 100644 index 00000000..d30323ff --- /dev/null +++ b/packages/node-runtime/src/ai/__tests__/cross-chat-agent.test.ts @@ -0,0 +1,117 @@ +import assert from 'node:assert/strict' +import { describe, it } from 'node:test' +import type { Api as PiApi, Model as PiModel } from '@earendil-works/pi-ai' +import type { AIChatManager } from '../chats' +import type { AgentStreamChunk } from '../agent/event-handler' +import { buildCrossChatSystemPrompt, runCrossChatAgent } from '../cross-chat-agent' + +describe('cross-chat agent prompt', () => { + it('locks the agent to dedicated tools and makes scope semantic rather than persistent', () => { + const prompt = buildCrossChatSystemPrompt('zh-CN') + for (const tool of [ + 'resolve_chat_entities', + 'search_messages_globally', + 'get_cross_chat_message_context', + 'get_cross_chat_overview', + ]) { + assert.match(prompt, new RegExp(tool)) + } + assert.match(prompt, /不构成永久锁定范围/) + assert.match(prompt, /交集、并集/) + assert.match(prompt, /唯一候选自动继续/) + assert.match(prompt, /多个候选必须停下来请用户确认/) + assert.match(prompt, /限定 scopes 时，可以不提供关键词/) + assert.match(prompt, /最近.*90 天/) + assert.match(prompt, /recent_days/) + assert.match(prompt, /sender.*owner/) + assert.match(prompt, /本人发言.*检索种子/) + assert.match(prompt, /coverage/) + assert.doesNotMatch(prompt, /可以使用.*execute_sql/) + }) +}) + +describe('cross-chat agent lifecycle', () => { + it('skips compression work when the request is already aborted', async () => { + const controller = new AbortController() + controller.abort() + const events: AgentStreamChunk[] = [] + let compressionReads = 0 + const aiChatManager = { + getLatestSummary() { + compressionReads++ + return null + }, + } as unknown as AIChatManager + const piModel = { + id: 'test-model', + name: 'Test model', + api: 'openai-completions', + provider: 'test', + } as unknown as PiModel + + await runCrossChatAgent({ + userMessage: '分析一下', + aiChatId: 'global-chat-1', + piModel, + apiKey: 'test-key', + tools: [], + aiChatManager, + abortSignal: controller.signal, + onEvent: (event) => events.push(event), + }) + + assert.equal(compressionReads, 0) + assert.deepEqual( + events.filter((event) => event.type === 'status').map((event) => event.status?.phase), + ['aborted'] + ) + assert.equal(events.at(-1)?.type, 'done') + }) + + it('finishes as aborted when cancellation happens after the initial check', async () => { + const controller = new AbortController() + const events: AgentStreamChunk[] = [] + const loggedErrors: unknown[] = [] + const aiChatManager = { + getHistoryForAgent() { + controller.abort() + return [] + }, + } as unknown as AIChatManager + const piModel = { + id: 'test-model', + name: 'Test model', + api: 'openai-completions', + provider: 'test', + } as unknown as PiModel + + await runCrossChatAgent({ + userMessage: '你觉得这个方案怎么样？', + aiChatId: 'global-chat-1', + historyLeafMessageId: null, + piModel, + apiKey: 'test-key', + tools: [], + aiChatManager, + abortSignal: controller.signal, + onEvent: (event) => events.push(event), + logger: { + info: () => undefined, + warn: () => undefined, + error: (_category, _message, error) => loggedErrors.push(error), + }, + }) + + const terminalStatuses = events + .filter((event) => event.type === 'status') + .map((event) => event.status?.phase) + .filter((phase) => phase === 'completed' || phase === 'aborted' || phase === 'error') + assert.deepEqual(terminalStatuses, ['aborted']) + assert.equal( + events.some((event) => event.type === 'error'), + false + ) + assert.equal(events.at(-1)?.type, 'done') + assert.deepEqual(loggedErrors, []) + }) +}) diff --git a/packages/node-runtime/src/ai/chats.ts b/packages/node-runtime/src/ai/chats.ts index 7f453812..1164d232 100644 --- a/packages/node-runtime/src/ai/chats.ts +++ b/packages/node-runtime/src/ai/chats.ts @@ -9,7 +9,7 @@ import Database from 'better-sqlite3' import * as fs from 'fs' import * as path from 'path' import type { ChartPayload, ChatEvidencePayload } from '@openchatlab/core' -import { DEFAULT_GENERAL_ASSISTANT_ID } from '@openchatlab/shared-types' +import { DEFAULT_GENERAL_ASSISTANT_ID, type AIEntityRef } from '@openchatlab/shared-types' import type { PlanContentBlock, PlanDraftContentBlock } from './agent' const DEFAULT_GENERAL_ID = DEFAULT_GENERAL_ASSISTANT_ID @@ -29,18 +29,7 @@ export interface AIChat { export type AIChatKind = 'session' | 'global' -export type AIEntityRef = - | { - type: 'contact' - contactKey: string - displayName: string - } - | { - type: 'session' - sessionId: string - displayName: string - sessionType: 'private' | 'group' - } +export type { AIEntityRef } from '@openchatlab/shared-types' export type ContentBlock = | { type: 'text'; text: string; processDurationMs?: number } diff --git a/packages/node-runtime/src/ai/compression/__tests__/compressor.test.ts b/packages/node-runtime/src/ai/compression/__tests__/compressor.test.ts index 704a8806..89d60109 100644 --- a/packages/node-runtime/src/ai/compression/__tests__/compressor.test.ts +++ b/packages/node-runtime/src/ai/compression/__tests__/compressor.test.ts @@ -190,4 +190,34 @@ describe('checkAndCompress tool result token accounting', () => { cleanup(dir) } }) + + it('does not persist a summary when cancellation happens during compression', async () => { + const dir = createTempDir() + const manager = createManager(dir) + const controller = new AbortController() + let receivedSignal: AbortSignal | undefined + try { + const chatId = seedToolHeavyChat(manager, true) + const adapter: CompressionLlmAdapter = { + contextWindow: 1000, + compress: async (_prompt, _maxTokens, signal) => { + receivedSignal = signal + controller.abort() + return 'SUMMARY THAT MUST NOT BE PERSISTED' + }, + } + + const result = await checkAndCompress(chatId, CONFIG, 'system', adapter, manager, undefined, { + signal: controller.signal, + }) + + assert.equal(receivedSignal, controller.signal) + assert.equal(result.compressed, false) + assert.equal(result.error, 'Compression aborted') + assert.equal(manager.getLatestSummary(chatId), null) + } finally { + manager.close() + cleanup(dir) + } + }) }) diff --git a/packages/node-runtime/src/ai/compression/adapter-factory.ts b/packages/node-runtime/src/ai/compression/adapter-factory.ts index 9d5e839d..06b46220 100644 --- a/packages/node-runtime/src/ai/compression/adapter-factory.ts +++ b/packages/node-runtime/src/ai/compression/adapter-factory.ts @@ -22,7 +22,7 @@ export function createCompressionLlmAdapter(options: CreateCompressionLlmAdapter return { contextWindow, - compress: async (prompt: string, maxTokens: number) => { + compress: async (prompt: string, maxTokens: number, signal?: AbortSignal) => { onCompressing?.() try { const result = await completeSimple( @@ -31,7 +31,7 @@ export function createCompressionLlmAdapter(options: CreateCompressionLlmAdapter systemPrompt: undefined, messages: [{ role: 'user', content: [{ type: 'text', text: prompt }], timestamp: Date.now() }] as any, }, - { apiKey, maxTokens } + { apiKey, maxTokens, signal } ) const text = result.content .filter((item): item is PiTextContent => item.type === 'text') @@ -39,6 +39,7 @@ export function createCompressionLlmAdapter(options: CreateCompressionLlmAdapter .join('') return text || null } catch (error) { + if (signal?.aborted) return null onError?.(error) return null } diff --git a/packages/node-runtime/src/ai/compression/compressor.ts b/packages/node-runtime/src/ai/compression/compressor.ts index ef3dd881..6f808c34 100644 --- a/packages/node-runtime/src/ai/compression/compressor.ts +++ b/packages/node-runtime/src/ai/compression/compressor.ts @@ -69,9 +69,11 @@ export async function checkAndCompress( systemPrompt: string, llmAdapter: CompressionLlmAdapter, convManager: AIChatManager, - logger: CompressionLogger = defaultLogger + logger: CompressionLogger = defaultLogger, + options: { signal?: AbortSignal } = {} ): Promise { try { + throwIfAborted(options.signal) const contextWindow = llmAdapter.contextWindow || DEFAULT_CONTEXT_WINDOW const thresholdTokens = Math.floor(contextWindow * (config.tokenThresholdPercent / 100) * 0.95) @@ -140,7 +142,8 @@ export async function checkAndCompress( const template = isProgressive ? PROGRESSIVE_COMPRESSION_PROMPT : INITIAL_COMPRESSION_PROMPT const prompt = template.replace('{maxTokens}', String(targetTokens)).replace('{messages}', compressInput) - let summaryText = await llmAdapter.compress(prompt, targetTokens) + let summaryText = await llmAdapter.compress(prompt, targetTokens, options.signal) + throwIfAborted(options.signal) if (!summaryText) { logger.warn('Compression', 'LLM compression failed, falling back to truncation') @@ -156,6 +159,7 @@ export async function checkAndCompress( summary?.entityRefs, ...messagesToCompress.map((message) => message.entityRefs) ) + throwIfAborted(options.signal) convManager.addSummaryMessage( aiChatId, summaryText, @@ -200,11 +204,21 @@ export async function checkAndCompress( summaryContent: summaryText, } } catch (error) { + if (options.signal?.aborted) { + return { compressed: false, reason: 'error', error: 'Compression aborted' } + } logger.error('Compression', 'Compression failed', { error: String(error) }) return { compressed: false, reason: 'error', error: String(error) } } } +function throwIfAborted(signal?: AbortSignal): void { + if (!signal?.aborted) return + const error = new Error('Compression aborted') + error.name = 'AbortError' + throw error +} + export async function manualCompress( aiChatId: string, config: CompressionConfig, diff --git a/packages/node-runtime/src/ai/compression/types.ts b/packages/node-runtime/src/ai/compression/types.ts index 38d13726..b4556912 100644 --- a/packages/node-runtime/src/ai/compression/types.ts +++ b/packages/node-runtime/src/ai/compression/types.ts @@ -36,7 +36,7 @@ export interface CompressionLlmAdapter { /** * 调用 LLM 进行压缩，返回压缩后的文本，失败返回 null。 */ - compress(prompt: string, maxTokens: number): Promise + compress(prompt: string, maxTokens: number, signal?: AbortSignal): Promise /** 解析模型的 context window 大小 */ contextWindow: number } diff --git a/packages/node-runtime/src/ai/cross-chat-agent.ts b/packages/node-runtime/src/ai/cross-chat-agent.ts new file mode 100644 index 00000000..99ad8de8 --- /dev/null +++ b/packages/node-runtime/src/ai/cross-chat-agent.ts @@ -0,0 +1,242 @@ +import type { AgentTool } from '@earendil-works/pi-agent-core' +import type { Api as PiApi, Message as PiMessage, Model as PiModel } from '@earendil-works/pi-ai' +import type { AIEntityRef } from '@openchatlab/shared-types' +import type { ThinkingLevel } from '@openchatlab/core' +import { DEFAULT_CONTEXT_COMPRESSION_CONFIG, checkAndCompress, createCompressionLlmAdapter } from './compression' +import { createAiTranslate } from './i18n' +import { initTokenizer } from './tokenizer' +import type { AIChatManager } from './chats' +import { AgentEventHandler, type AgentStreamChunk } from './agent/event-handler' +import { appendEntityRefsForModel } from './agent/history' +import { DEFAULT_MAX_TOOL_ROUNDS } from './agent/constants' +import { runAgentCore } from './agent/core' +import { buildPlanGuidance, createAnalysisPlanner, createPlanContentBlock } from './agent/planner' +import { createLlmRouteDecider, decideRequestRoute } from './agent/router' +import { formatAIError } from './error-formatter' + +export interface CrossChatAgentLogger { + info(category: string, message: string, data?: unknown): void + warn(category: string, message: string, data?: unknown): void + error(category: string, message: string, data?: unknown): void +} + +export interface RunCrossChatAgentOptions { + userMessage: string + entityRefs?: AIEntityRef[] + aiChatId: string + historyLeafMessageId?: string | null + locale?: string + piModel: PiModel + apiKey: string + tools: AgentTool[] + aiChatManager: AIChatManager + onEvent: (event: AgentStreamChunk) => void + abortSignal?: AbortSignal + thinkingLevel?: ThinkingLevel + logger?: CrossChatAgentLogger +} + +export async function runCrossChatAgent(options: RunCrossChatAgentOptions): Promise { + const { + userMessage, + entityRefs, + aiChatId, + historyLeafMessageId, + locale = 'zh-CN', + piModel, + apiKey, + tools, + aiChatManager, + onEvent, + abortSignal, + thinkingLevel, + logger, + } = options + await initTokenizer() + logger?.info('CrossChatAgent', 'Cross-chat agent execution started', { + aiChatId, + entityRefCount: entityRefs?.length ?? 0, + toolCount: tools.length, + }) + const systemPrompt = buildCrossChatSystemPrompt(locale) + const handler = new AgentEventHandler({ onChunk: onEvent, context: {}, systemPrompt }) + let cachedMessages: PiMessage[] = [] + + const finishAborted = () => { + handler.emitStatus('aborted', cachedMessages, { force: true }) + onEvent({ type: 'done', isFinished: true, usage: handler.cloneUsage() }) + } + + if (abortSignal?.aborted) { + finishAborted() + return + } + + if (historyLeafMessageId === undefined) { + const compressionResult = await checkAndCompress( + aiChatId, + DEFAULT_CONTEXT_COMPRESSION_CONFIG, + systemPrompt, + createCompressionLlmAdapter({ + piModel, + apiKey, + onCompressing: () => handler.emitStatus('compressing', []), + }), + aiChatManager, + logger, + { signal: abortSignal } + ) + if (compressionResult.compressed) { + onEvent({ + type: 'compression_done', + compressionResult: { + summaryContent: compressionResult.summaryContent ?? '', + tokensBefore: compressionResult.tokensBefore ?? 0, + tokensAfter: compressionResult.tokensAfter ?? 0, + timestamp: Date.now(), + }, + }) + } + } + + if (abortSignal?.aborted) { + finishAborted() + return + } + + const history = aiChatManager.getHistoryForAgent(aiChatId, undefined, historyLeafMessageId) + const modelUserMessage = appendEntityRefsForModel(userMessage, entityRefs) + handler.emitStatus('preparing', [], { pendingUserMessage: modelUserMessage, force: true }) + + try { + const routeInput = { + userMessage: modelUserMessage, + chatType: 'group' as const, + locale, + availableTools: tools.map((tool) => tool.name), + availableCapabilities: [], + } + const routeDecision = await decideRequestRoute(routeInput, { + llmRouter: createLlmRouteDecider({ piModel, apiKey, abortSignal }), + }) + onEvent({ type: 'route', routeDecision }) + + let effectiveSystemPrompt = systemPrompt + if (routeDecision.route === 'planned_execution') { + const planner = createAnalysisPlanner({ + piModel, + apiKey, + onPlanDelta: (delta) => onEvent({ type: 'plan_delta', planDelta: delta }), + onThinkingDelta: (delta) => onEvent({ type: 'think', content: delta, thinkTag: 'thinking' }), + onThinkingEnd: (durationMs) => + onEvent({ type: 'think', content: '', thinkTag: 'thinking', thinkDurationMs: durationMs }), + onValidationDelta: (delta) => onEvent({ type: 'think', content: delta, thinkTag: 'plan_validation' }), + onValidationEnd: (durationMs) => + onEvent({ type: 'think', content: '', thinkTag: 'plan_validation', thinkDurationMs: durationMs }), + }) + const plan = await planner(routeInput, abortSignal) + if (plan) { + onEvent({ type: 'plan', plan: createPlanContentBlock(plan) }) + effectiveSystemPrompt = `${systemPrompt}\n\n${buildPlanGuidance(plan)}` + } else { + onEvent({ type: 'plan_skipped' }) + } + } + + const result = await runAgentCore({ + piModel, + apiKey, + systemPrompt: effectiveSystemPrompt, + tools, + history, + userMessage: modelUserMessage, + maxToolRounds: DEFAULT_MAX_TOOL_ROUNDS, + abortSignal, + providerSessionId: aiChatId, + steerMessage: createAiTranslate(locale)('ai.agent.answerWithoutTools'), + thinkingLevel, + onConvertToLlm: (messages) => { + cachedMessages = messages as PiMessage[] + }, + onEvent: (event) => handler.handleCoreEvent(event, cachedMessages), + onDebugContext: (messages) => { + try { + aiChatManager.setPendingDebugContext(aiChatId, JSON.stringify(messages, null, 2)) + } catch { + // Debug context is best-effort. + } + }, + }) + if (abortSignal?.aborted) { + finishAborted() + return + } + if (result.error) { + onEvent({ type: 'error', error: { name: 'AgentError', message: formatAIError(result.error) } }) + } + handler.emitStatus('completed', cachedMessages, { force: true }) + logger?.info('CrossChatAgent', 'Cross-chat agent execution completed', { + aiChatId, + toolRounds: result.toolRounds, + toolsUsed: result.toolsUsed.length, + }) + onEvent({ type: 'done', isFinished: true, usage: result.usage }) + } catch (error) { + if (abortSignal?.aborted) { + finishAborted() + return + } + logger?.error('CrossChatAgent', 'Cross-chat agent execution failed', error) + handler.emitStatus('error', cachedMessages, { force: true }) + onEvent({ type: 'error', error: { name: 'AgentError', message: formatAIError(error) } }) + onEvent({ type: 'done', isFinished: true, usage: handler.cloneUsage() }) + } +} + +export function buildCrossChatSystemPrompt(locale = 'zh-CN'): string { + if (locale.startsWith('zh')) { + return `你是 ChatLab 的跨对话分析助手。你可以按用户当前问题，在其本地聊天数据库中按需检索多个联系人和群聊。 + +数据与范围规则： +- 用户已授权你查询全部本地聊天数据，但只能为回答当前问题按需调用工具，禁止无目的遍历。 +- 是界面选择器写入的稳定实体引用，涉及这些实体时直接调用 resolve_chat_entities。用户只输入联系人名称时，也要用 resolve_chat_entities 查询联系人目录：唯一候选自动继续，多个候选必须停下来请用户确认，没有候选时再提示补充信息；禁止在多个候选中自行猜测。 +- 对话历史中的实体引用只帮助理解上下文，不构成永久锁定范围。每一轮根据用户语义决定继续原对象、切换对象或执行全局发现。 +- 联系人默认覆盖其实际参与的私聊和群聊。多个对象既可能是交集、并集，也可能需要分别检索比较；以用户语义为准，不要机械套用一种集合规则。 +- 只有用户明确表达“忘了和谁聊过”“在所有聊天里找”等全局发现意图时，才允许不带 scopes 调用 search_messages_globally。全局发现必须提供至少一个关键词；已经限定 scopes 时，可以不提供关键词来抽取近期消息样本。 +- 用户使用“最近”“近期”等相对时间但没有给出具体范围时，第一次搜索必须传 recent_days=90，并在回答中说明按最近 90 天统计；禁止先扫描全部历史再事后主观截取。只有 90 天内没有结果时，才能询问用户是否扩大范围。 +- 用户把自己作为事件主体（例如“我最近跟多少人聊过我买房”）时，第一次搜索必须传 sender=owner，把本人发言作为检索种子；不要先搜索所有人的同类话题。命中后再用 get_cross_chat_message_context 展开周边消息，识别真正参与对话的人；上下文不限制为本人发言。 + +工具与结论规则： +- 你只有 resolve_chat_entities、search_messages_globally、get_cross_chat_message_context、get_cross_chat_overview 四个工具。不要声称可以使用单会话 SQL、语义索引、技能、图表或热力图。 +- search_messages_globally 提供关键词时是字面 LIKE 检索，不是语义搜索。用户问“聊过什么”等开放问题时，先限定联系人或群聊 scopes，再无关键词抽取近期消息并按需展开上下文；不要对全部会话做无关键词扫描。 +- 比较联系人或群聊时，优先分别给出概览并检索有来源的证据。需要理解命中消息时，用 session_id + message_id 获取上下文。 +- 工具返回 coverage 和 truncated。覆盖不完整或被截断时必须明确说明抽样范围，禁止把样本表述成全量结论。 +- sender=owner 时还要检查 coverage.ownerResolution；存在未设置或无法解析“我”的会话时，必须说明对应覆盖缺口，禁止通过昵称猜测本人。 +- 引用证据时说明来源会话；不要泄露工具未返回的信息，也不要编造联系人、群聊或消息。 +- 如果问题不需要聊天数据，直接回答；如果现有四个工具不足，坦诚说明第一版能力边界。 + +你可以使用工具。如果需要你没有的信息，请调用提供的函数。` + } + + return `You are ChatLab's cross-chat analysis assistant. You may query multiple contacts and group chats from the user's local chat databases as needed for the current question. + +Data and scope rules: +- The user authorizes access to all local chat data, but you must query only what is needed for the current question and never crawl without purpose. +- contains stable references selected in the UI and should be passed directly to resolve_chat_entities. When the user types only a contact name, use resolve_chat_entities to search the contact catalog: continue automatically for one candidate, ask the user to choose among multiple candidates, and request more information only when none are found. Never guess among ambiguous candidates. +- Entity references in history provide conversational context, not a permanently locked scope. Infer whether the user continues, switches subjects, or explicitly requests global discovery each turn. +- A contact normally covers the private and group sessions they actually participate in. Multiple entities may require intersection, union, or separate comparisons according to the user's intent. +- Call search_messages_globally without scopes only for explicit global discovery such as "I forgot who I discussed this with". Global discovery always requires at least one keyword; scoped searches may omit keywords to sample recent messages. +- When the user says "recent" or "recently" without a specific range, the first search must pass recent_days=90 and the answer must state that it covers the last 90 days. Never scan all history first and apply a subjective cutoff afterward. Ask before widening only when the 90-day search finds nothing. +- When the user is the subject of the event, such as "who did I discuss my home purchase with", the first search must pass sender=owner and treat the owner's messages as discovery seeds. Then use get_cross_chat_message_context to identify actual participants; surrounding context is not owner-only. + +Tool and conclusion rules: +- You only have resolve_chat_entities, search_messages_globally, get_cross_chat_message_context, and get_cross_chat_overview. Do not claim access to session SQL, semantic search, skills, charts, or heatmaps. +- Search with keywords is literal LIKE matching, not semantic retrieval. For open-ended questions such as "what did we discuss", first resolve contact or session scopes, then sample recent messages without keywords and expand relevant context. Never scan every session without keywords. +- For comparisons, prefer separate overviews and source-backed evidence. Expand a hit with session_id plus message_id when context is needed. +- Respect coverage and truncated fields. State incomplete coverage or sampling explicitly and never present a sample as exhaustive. +- For sender=owner, inspect coverage.ownerResolution and disclose sessions where the owner is missing or unresolved. Never guess the owner from display names. +- Name the source session when citing evidence. Never invent people, sessions, or messages. +- Answer directly when no chat data is needed. If the four tools are insufficient, explain the first-version limitation honestly. + +You have access to tools. If you need information you don't have, use the provided functions.` +} diff --git a/packages/node-runtime/src/ai/index.ts b/packages/node-runtime/src/ai/index.ts index d184b5ba..89980cce 100644 --- a/packages/node-runtime/src/ai/index.ts +++ b/packages/node-runtime/src/ai/index.ts @@ -139,6 +139,8 @@ export type { LlmRouteDecider, RequestRoute, RouteDecision, RouteDecisionSource, export { buildPlanGuidance, createAnalysisPlanner, createPlanContentBlock } from './agent' export { createDataSnapshotFromOverview } from './agent' export { buildSemanticSearchGuidance } from './agent' +export { buildCrossChatSystemPrompt, runCrossChatAgent } from './cross-chat-agent' +export type { CrossChatAgentLogger, RunCrossChatAgentOptions } from './cross-chat-agent' export type { AnalysisPlanIntent, AnalysisPlanner, diff --git a/packages/node-runtime/src/index.ts b/packages/node-runtime/src/index.ts index 0134391f..93aebc69 100644 --- a/packages/node-runtime/src/index.ts +++ b/packages/node-runtime/src/index.ts @@ -310,6 +310,8 @@ export { DEFAULT_MAX_TOOL_ROUNDS, createLlmRouteDecider, decideRequestRoute, run export type { LlmRouteDecider, RequestRoute, RouteDecision, RouteDecisionSource, RouterInput } from './ai' export { buildPlanGuidance, createAnalysisPlanner, createDataSnapshotFromOverview, createPlanContentBlock } from './ai' export { buildSemanticSearchGuidance } from './ai' +export { buildCrossChatSystemPrompt, runCrossChatAgent } from './ai' +export type { CrossChatAgentLogger, RunCrossChatAgentOptions } from './ai' export type { AnalysisPlanIntent, AnalysisPlanner, @@ -495,6 +497,7 @@ export { MergeSessionCache, createContactsService, createCrossChatAnalysisService, + preprocessCrossChatMessages, createPeopleRelationshipsService, createGlobalInsightService, assertValidTopicDayKey, diff --git a/packages/node-runtime/src/services/cross-chat-analysis/index.ts b/packages/node-runtime/src/services/cross-chat-analysis/index.ts index 9cee27fb..87db8cf2 100644 --- a/packages/node-runtime/src/services/cross-chat-analysis/index.ts +++ b/packages/node-runtime/src/services/cross-chat-analysis/index.ts @@ -1,4 +1,5 @@ export { createCrossChatAnalysisService } from './service' +export { preprocessCrossChatMessages } from './preprocess' export type { CrossChatAnalysisService, CrossChatAnalysisServiceDeps } from './service' export type { CrossChatEntityResolution, diff --git a/packages/node-runtime/src/services/cross-chat-analysis/preprocess.ts b/packages/node-runtime/src/services/cross-chat-analysis/preprocess.ts new file mode 100644 index 00000000..ee1b23aa --- /dev/null +++ b/packages/node-runtime/src/services/cross-chat-analysis/preprocess.ts @@ -0,0 +1,37 @@ +import { getSessionMeta } from '@openchatlab/core' +import type { CrossChatMessageSource } from '@openchatlab/shared-types' +import type { PreprocessConfig } from '../../ai/preprocessor' +import { preprocessMessages } from '../../ai/preprocessor' +import type { SessionRuntimeAdapter } from '../adapters' + +/** + * Apply message cleaning per source session before cross-chat evidence reaches an LLM. + * Session-qualified pseudonyms prevent unrelated local member IDs from being conflated across databases. + */ +export function preprocessCrossChatMessages( + adapter: SessionRuntimeAdapter, + sessionId: string, + messages: CrossChatMessageSource[], + config?: PreprocessConfig +): CrossChatMessageSource[] { + const safeConfig = config ? { ...config, mergeConsecutive: false, anonymizeNames: false } : undefined + const processed = preprocessMessages( + messages.map((message) => ({ ...message })), + safeConfig + ) + if (!config?.anonymizeNames) return processed + + let ownerPlatformId: string | null = null + try { + const db = adapter.openReadonly(sessionId) + ownerPlatformId = db ? getSessionMeta(db)?.ownerId?.trim() || null : null + } catch { + // Anonymization remains safe without owner labeling. + } + + return processed.map((message) => ({ + ...message, + senderName: + ownerPlatformId && message.senderPlatformId === ownerPlatformId ? 'Owner' : `U${message.senderId}@${sessionId}`, + })) +} 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 index 10cd0a2e..9c0e9b60 100644 --- a/packages/node-runtime/src/services/cross-chat-analysis/service.test.ts +++ b/packages/node-runtime/src/services/cross-chat-analysis/service.test.ts @@ -3,13 +3,19 @@ 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 { CHAT_DB_SCHEMA, generateSessionIndex } from '@openchatlab/core' import type { DatabaseAdapter } from '@openchatlab/core' -import type { ContactDetailResponse, ContactItem } from '@openchatlab/shared-types' +import { + ChatType, + type ContactDetailResponse, + type ContactItem, + type ContactsResponse, +} from '@openchatlab/shared-types' import { openBetterSqliteDatabase } from '../../better-sqlite3-adapter' import type { ContactsService } from '../contacts' import type { SessionRuntimeAdapter } from '../adapters' import { createCrossChatAnalysisService } from './service' +import { preprocessCrossChatMessages } from './preprocess' const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node') @@ -17,6 +23,7 @@ interface SeedSession { id: string name: string type: 'private' | 'group' + ownerPlatformId?: string members: Array<{ id: number; platformId: string; name: string }> messages: Array<{ id: number; senderId: number; ts: number; content: string }> } @@ -61,11 +68,12 @@ class TestEnvironment { 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( + db.prepare('INSERT INTO meta (name, platform, type, imported_at, owner_id) VALUES (?, ?, ?, ?, ?)').run( session.name, 'test', session.type, - 1780000000 + 1780000000, + session.ownerPlatformId ?? null ) for (const member of session.members) { db.prepare('INSERT INTO member (id, platform_id, account_name) VALUES (?, ?, ?)').run( @@ -131,7 +139,7 @@ function detail( function createFixture(): { env: TestEnvironment - contactsService: Pick + contactsService: Pick } { const env = new TestEnvironment() env.seed({ @@ -170,43 +178,139 @@ function createFixture(): { messages: [{ id: 1, senderId: 30, ts: 200, content: 'project alpha from another Alice' }], }) + const contactItems = [ + contact({ + key: 'test:alice', + platformId: 'alice', + displayName: 'Alice', + aliases: ['Ally'], + sourceSessions: [ + { id: 'private-alice', name: 'Alice private', platform: 'test', type: ChatType.PRIVATE }, + { id: 'group-work', name: 'Work group', platform: 'test', type: ChatType.GROUP }, + ], + }), + 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: ChatType.GROUP }], + }), + ] 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' }], - }) - ), - ], + ['test:alice', detail(contactItems[0])], + ['test:group-other:alice-other', detail(contactItems[1])], ]) return { env, contactsService: { getContactDetail: (key) => contacts.get(key) ?? detail(null), + getContactsPage: (options = {}) => { + const query = options.query?.trim().toLocaleLowerCase() ?? '' + const matches = contactItems.filter((item) => { + const values = [item.displayName, ...item.aliases].map((value) => value.toLocaleLowerCase()) + return !query || values.some((value) => value.includes(query)) + }) + return { + contacts: matches.map(({ sourceSessions: _sourceSessions, searchText: _searchText, ...item }) => item), + cache: { status: 'fresh', computedAt: 1780000000 }, + pagination: { page: 1, pageSize: 100, total: matches.length, hasMore: false }, + task: { id: null, status: 'idle', startedAt: null, finishedAt: null, processedSessions: 0, totalSessions: 0 }, + } as ContactsResponse + }, }, } } +test('contact lookup resolves a unique alias and preserves same-name ambiguity', () => { + const { env, contactsService } = createFixture() + try { + const service = createCrossChatAnalysisService({ adapter: env.adapter, contactsService }) + assert.deepEqual(service.lookupContact('Ally'), { + query: 'Ally', + status: 'resolved', + cacheStatus: 'fresh', + totalCandidates: 1, + candidates: [ + { + contactKey: 'test:alice', + displayName: 'Alice', + platform: 'test', + aliases: ['Ally'], + sourceSessions: [ + { id: 'private-alice', name: 'Alice private', type: ChatType.PRIVATE }, + { id: 'group-work', name: 'Work group', type: ChatType.GROUP }, + ], + }, + ], + }) + const ambiguous = service.lookupContact('Alice') + assert.equal(ambiguous.status, 'ambiguous') + assert.equal(ambiguous.totalCandidates, 2) + } finally { + env.cleanup() + } +}) + +test('contact lookup uses the all-history snapshot for typed names and candidate details', () => { + const { env } = createFixture() + const pagePresets: Array = [] + const detailPresets: Array = [] + const legacyContact = contact({ + key: 'test:legacy', + platformId: 'alice', + displayName: 'Legacy Alice', + sourceSessions: [{ id: 'private-alice', name: 'Alice private', platform: 'test', type: ChatType.PRIVATE }], + }) + const { sourceSessions: _sourceSessions, searchText: _searchText, ...legacyListContact } = legacyContact + const contactsService: Pick = { + getContactsPage: (options = {}) => { + pagePresets.push(options.timeRangePreset) + const contacts = options.timeRangePreset === 'all' ? [legacyListContact] : [] + return { + contacts, + diagnostics: { + privateSessionCount: 1, + activePrivateSessionCount: 1, + contactsEnabled: true, + skippedMissingOwnerSessions: 0, + skippedUnresolvedOwnerSessions: 0, + skippedAmbiguousPrivateSessions: 0, + skippedInvalidPlatformIdMembers: 0, + skippedFailedSessions: 0, + warnings: [], + }, + cache: { status: 'fresh', computedAt: 1780000000 }, + timeRange: { preset: 'all', anchorTs: null, startTs: null }, + algorithmVersion: 'test', + pagination: { page: 1, pageSize: 200, total: contacts.length, hasMore: false }, + stats: { friendsTotal: contacts.length, nonFriendsTotal: 0 }, + task: { id: null, status: 'idle', startedAt: null, finishedAt: null, processedSessions: 0, totalSessions: 0 }, + } + }, + getContactDetail: (_key, options) => { + detailPresets.push(options?.timeRangePreset) + return detail(options?.timeRangePreset === 'all' ? legacyContact : null) + }, + } + + try { + const service = createCrossChatAnalysisService({ adapter: env.adapter, contactsService }) + const result = service.lookupContact('Legacy Alice') + + assert.equal(result.status, 'resolved') + assert.deepEqual( + result.candidates[0]?.sourceSessions.map((session) => session.id), + ['private-alice'] + ) + assert.deepEqual(pagePresets, ['all']) + assert.deepEqual(detailPresets, ['all']) + } finally { + env.cleanup() + } +}) + test('entity resolution uses contact keys and resolves per-session member ids without merging display names', () => { const { env, contactsService } = createFixture() try { @@ -240,9 +344,10 @@ test('entity resolution uses contact keys and resolves per-session member ids wi }) test('entity resolution uses the all-history contact snapshot so older source sessions remain searchable', () => { - const { env } = createFixture() + const { env, contactsService: fixtureContactsService } = createFixture() try { - const contactsService: Pick = { + const contactsService: Pick = { + ...fixtureContactsService, getContactDetail: (_key, options) => detail( contact({ @@ -252,10 +357,10 @@ test('entity resolution uses the all-history contact snapshot so older source se sourceSessions: options?.timeRangePreset === 'all' ? [ - { id: 'private-alice', name: 'Alice private', platform: 'test', type: 'private' }, - { id: 'group-work', name: 'Work group', platform: 'test', type: 'group' }, + { id: 'private-alice', name: 'Alice private', platform: 'test', type: ChatType.PRIVATE }, + { id: 'group-work', name: 'Work group', platform: 'test', type: ChatType.GROUP }, ] - : [{ id: 'group-work', name: 'Work group', platform: 'test', type: 'group' }], + : [{ id: 'group-work', name: 'Work group', platform: 'test', type: ChatType.GROUP }], }) ), } @@ -309,6 +414,40 @@ test('scoped search filters by resolved member ids and keeps compound evidence i } }) +test('message context stays inside the indexed segment around the matched message', () => { + const { env, contactsService } = createFixture() + try { + env.seed({ + id: 'segmented-chat', + name: 'Segmented chat', + type: 'group', + members: [ + { id: 1, platformId: 'owner', name: 'Me' }, + { id: 2, platformId: 'other', name: 'Other' }, + ], + messages: [ + { id: 1, senderId: 1, ts: 100, content: 'first segment start' }, + { id: 2, senderId: 2, ts: 110, content: 'first segment end' }, + { id: 3, senderId: 1, ts: 1_000, content: 'matched second segment start' }, + { id: 4, senderId: 2, ts: 1_010, content: 'second segment continuation' }, + ], + }) + const db = env.adapter.ensureWritable('segmented-chat') + generateSessionIndex(db, 100) + db.close() + const service = createCrossChatAnalysisService({ adapter: env.adapter, contactsService }) + + const context = service.getMessageContext({ sessionId: 'segmented-chat', messageId: 3, contextSize: 1 }) + + assert.deepEqual( + context.messages.map((message) => message.messageId), + [3, 4] + ) + } finally { + env.cleanup() + } +}) + test('global search scans recent sessions first and reports budget truncation', async () => { const { env, contactsService } = createFixture() try { @@ -333,11 +472,82 @@ test('global search scans recent sessions first and reports budget truncation', } }) -test('search rejects empty keywords and honors interruption before scanning', async () => { +test('global search applies the relative time range and resolves the owner independently in each session', async () => { + const { env, contactsService } = createFixture() + const nowSeconds = 10_000_000 + try { + env.seed({ + id: 'recent-owner-session', + name: 'Recent owner session', + type: 'group', + ownerPlatformId: 'owner-local', + members: [ + { id: 40, platformId: 'owner-local', name: 'Me' }, + { id: 41, platformId: 'other', name: 'Other' }, + ], + messages: [ + { id: 1, senderId: 40, ts: nowSeconds - 100, content: 'buying a home' }, + { id: 2, senderId: 41, ts: nowSeconds - 90, content: 'buying a home too' }, + { id: 3, senderId: 40, ts: nowSeconds - 91 * 86400, content: 'old buying a home note' }, + ], + }) + env.seed({ + id: 'missing-owner-session', + name: 'Missing owner session', + type: 'private', + members: [{ id: 50, platformId: 'someone', name: 'Someone' }], + messages: [{ id: 1, senderId: 50, ts: nowSeconds - 80, content: 'buying a home' }], + }) + const service = createCrossChatAnalysisService({ + adapter: env.adapter, + contactsService, + now: () => nowSeconds * 1000, + }) + const result = await service.searchMessages({ + keywords: ['buying a home'], + recentDays: 90, + sender: 'owner', + maxSessions: 10, + maxEvidence: 10, + }) + + assert.deepEqual( + result.messages.map((message) => [message.sessionId, message.messageId, message.senderId]), + [['recent-owner-session', 1, 40]] + ) + assert.deepEqual(result.appliedFilters, { + startTs: nowSeconds - 90 * 86400, + endTs: nowSeconds, + recentDays: 90, + sender: 'owner', + }) + assert.deepEqual(result.coverage.ownerResolution, { + resolvedSessions: 1, + missingOwnerSessions: 4, + unresolvedOwnerSessions: 0, + }) + } finally { + env.cleanup() + } +}) + +test('search allows empty keywords for scoped sampling but rejects unscoped scans', async () => { const { env, contactsService } = createFixture() try { const service = createCrossChatAnalysisService({ adapter: env.adapter, contactsService }) await assert.rejects(() => service.searchMessages({ keywords: [] }), /keyword/i) + const scoped = await service.searchMessages({ + keywords: [], + scopes: [{ sessionId: 'group-work', memberIds: [20] }], + maxEvidence: 10, + }) + assert.deepEqual( + scoped.messages.map((message) => [message.sessionId, message.messageId, message.senderId]), + [ + ['group-work', 3, 20], + ['group-work', 1, 20], + ] + ) const controller = new AbortController() controller.abort() @@ -396,14 +606,14 @@ test('search stops between sessions when the wall-time budget is exhausted', asy } }) -test('overview returns separate scoped totals instead of combining unrelated senders', async () => { +test('overview preserves separate member scopes for contacts in the same group', 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' }, + { sessionId: 'group-work', memberIds: [21], label: 'Bob in Work group' }, ], }) @@ -411,10 +621,55 @@ test('overview returns separate scoped totals instead of combining unrelated sen result.items.map((item) => [item.label, item.totalMessages, item.firstMessageTs, item.lastMessageTs]), [ ['Alice in Work group', 2, 300, 320], - ['Other Alice', 1, 200, 200], + ['Bob in Work group', 1, 310, 310], ] ) } finally { env.cleanup() } }) + +test('cross-chat anonymization namespaces local member ids by source session', () => { + const { env } = createFixture() + try { + const base = { + sessionName: 'Session', + sessionType: ChatType.GROUP, + platform: 'test', + lastMessageTs: 1, + messageId: 1, + senderId: 10, + senderName: 'Alice', + senderPlatformId: 'alice', + content: 'hello', + timestamp: 1, + messageType: 0, + } + const config = { + dataCleaning: false, + mergeConsecutive: true, + blacklistKeywords: [], + denoise: false, + desensitize: false, + desensitizeRules: [], + anonymizeNames: true, + } + const first = preprocessCrossChatMessages( + env.adapter, + 'private-alice', + [{ ...base, sessionId: 'private-alice' }], + config + ) + const second = preprocessCrossChatMessages( + env.adapter, + 'group-work', + [{ ...base, sessionId: 'group-work' }], + config + ) + + assert.equal(first[0].senderName, 'U10@private-alice') + assert.equal(second[0].senderName, 'U10@group-work') + } 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 index e8c2d85e..edd823a5 100644 --- a/packages/node-runtime/src/services/cross-chat-analysis/service.ts +++ b/packages/node-runtime/src/services/cross-chat-analysis/service.ts @@ -1,35 +1,37 @@ import { getMembers, - getMessageContext as getCoreMessageContext, getRecentMessages, + getSearchMessageContext as getCoreSearchMessageContext, getSessionMeta, getSessionOverview, searchMessagesByKeywords, type DatabaseAdapter, } from '@openchatlab/core' -import { ChatType } from '@openchatlab/shared-types' -import type { AIEntityRef } from '../../ai/chats' +import { + ChatType, + type AIEntityRef, + type CrossChatContactCandidate, + type CrossChatContactLookupResult, + type CrossChatEntityResolution, + type CrossChatMessageContextRequest, + type CrossChatMessageContextResult, + type CrossChatMessageSource, + type CrossChatOperationOptions, + type CrossChatOverviewItem, + type CrossChatOverviewRequest, + type CrossChatOverviewResult, + type CrossChatResolvedContact, + type CrossChatResolvedSession, + type CrossChatSearchRequest, + type CrossChatSearchResult, + type CrossChatSearchScope, + type CrossChatSessionDescriptor, + type CrossChatTruncationReason, + type CrossChatUnresolvedEntity, +} from '@openchatlab/shared-types' 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 @@ -39,14 +41,17 @@ 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 +const SECONDS_PER_DAY = 86400 +const MAX_RECENT_DAYS = 3650 export interface CrossChatAnalysisServiceDeps { adapter: SessionRuntimeAdapter - contactsService: Pick + contactsService: Pick now?: () => number } export interface CrossChatAnalysisService { + lookupContact(query: string): CrossChatContactLookupResult resolveEntities(refs: AIEntityRef[]): CrossChatEntityResolution searchMessages(request: CrossChatSearchRequest, options?: CrossChatOperationOptions): Promise getMessageContext(request: CrossChatMessageContextRequest): CrossChatMessageContextResult @@ -60,6 +65,56 @@ export function createCrossChatAnalysisService(deps: CrossChatAnalysisServiceDep class DefaultCrossChatAnalysisService implements CrossChatAnalysisService { constructor(private readonly deps: CrossChatAnalysisServiceDeps) {} + lookupContact(rawQuery: string): CrossChatContactLookupResult { + const query = rawQuery.trim() + const response = this.deps.contactsService.getContactsPage({ + acceptStale: true, + page: 1, + pageSize: 200, + query, + timeRangePreset: 'all', + }) + const normalizedQuery = normalizeContactName(query) + const exactMatches = response.contacts.filter((contact) => + [contact.displayName, ...contact.aliases].some((name) => normalizeContactName(name) === normalizedQuery) + ) + const matchedContacts = exactMatches.length > 0 ? exactMatches : response.contacts + const totalCandidates = exactMatches.length > 0 ? exactMatches.length : response.pagination.total + const candidates = matchedContacts.slice(0, 8).map((contact): CrossChatContactCandidate => { + const detail = this.deps.contactsService.getContactDetail(contact.key, { + acceptStale: true, + timeRangePreset: 'all', + }) + return { + contactKey: contact.key, + displayName: contact.displayName, + platform: contact.platform, + aliases: contact.aliases, + sourceSessions: + detail.contact?.sourceSessions.map((session) => ({ + id: session.id, + name: session.name, + type: session.type, + })) ?? [], + } + }) + const status: CrossChatContactLookupResult['status'] = + totalCandidates === 1 + ? 'resolved' + : totalCandidates > 1 + ? 'ambiguous' + : response.cache.status === 'missing' || response.task?.status === 'running' + ? 'unavailable' + : 'not_found' + return { + query, + status, + cacheStatus: response.cache.status, + totalCandidates, + candidates, + } + } + resolveEntities(refs: AIEntityRef[]): CrossChatEntityResolution { const contacts: CrossChatResolvedContact[] = [] const sessions: CrossChatResolvedSession[] = [] @@ -178,15 +233,34 @@ class DefaultCrossChatAnalysisService implements CrossChatAnalysisService { 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') + if (keywords.length === 0 && (!request.scopes || request.scopes.length === 0)) { + throw new Error('At least one search keyword is required for an unscoped search') + } throwIfAborted(options.signal) + const sender = request.sender === 'owner' ? 'owner' : 'all' + const recentDays = normalizeRecentDays(request.recentDays) + const effectiveRecentDays = request.startTs === undefined ? recentDays : undefined + let startTs = request.startTs + let endTs = request.endTs + if (effectiveRecentDays !== undefined) { + endTs ??= Math.floor(this.now() / 1000) + startTs = endTs - effectiveRecentDays * SECONDS_PER_DAY + } + const ownerResolution = + sender === 'owner' ? { resolvedSessions: 0, missingOwnerSessions: 0, unresolvedOwnerSessions: 0 } : undefined + 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 { candidates, candidateSessionCount } = this.resolveSearchCandidates( + request.scopes, + failedSessionIds, + sender, + ownerResolution + ) const selected = candidates.slice(0, maxSessions) const truncatedReasons = new Set() if (candidates.length > selected.length) truncatedReasons.add('session_budget') @@ -218,8 +292,8 @@ class DefaultCrossChatAnalysisService implements CrossChatAnalysisService { continue } const result = searchMessagesByKeywords(db, keywords, { - startTs: request.startTs, - endTs: request.endTs, + startTs, + endTs, senderIds: candidate.memberIds, matchMode: request.matchMode, sort: request.sort, @@ -255,11 +329,18 @@ class DefaultCrossChatAnalysisService implements CrossChatAnalysisService { return { messages, totalMatches, + appliedFilters: { + startTs: startTs ?? null, + endTs: endTs ?? null, + recentDays: effectiveRecentDays ?? null, + sender, + }, coverage: { candidateSessions: candidateSessionCount, scannedSessions, matchedSessions, failedSessions: failedSessionIds.size, + ownerResolution, truncated: truncatedReasons.size > 0, truncatedReasons: [...truncatedReasons], }, @@ -271,7 +352,7 @@ class DefaultCrossChatAnalysisService implements CrossChatAnalysisService { 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) + const messages = getCoreSearchMessageContext(db, [request.messageId], contextSize, contextSize) if (!messages.some((message) => message.id === request.messageId)) { throw createNotFoundError(`Message not found: ${request.messageId}`) } @@ -286,7 +367,7 @@ class DefaultCrossChatAnalysisService implements CrossChatAnalysisService { options: CrossChatOperationOptions = {} ): Promise { throwIfAborted(options.signal) - const scopes = normalizeScopes(request.scopes) + const scopes = normalizeOverviewScopes(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) @@ -345,7 +426,13 @@ class DefaultCrossChatAnalysisService implements CrossChatAnalysisService { private resolveSearchCandidates( scopes: CrossChatSearchScope[] | undefined, - failedSessionIds: Set + failedSessionIds: Set, + sender: 'all' | 'owner', + ownerResolution?: { + resolvedSessions: number + missingOwnerSessions: number + unresolvedOwnerSessions: number + } ): { candidates: SearchCandidate[]; candidateSessionCount: number } { const normalizedScopes: CrossChatSearchScope[] = scopes ? normalizeScopes(scopes) @@ -357,6 +444,26 @@ class DefaultCrossChatAnalysisService implements CrossChatAnalysisService { failedSessionIds.add(scope.sessionId) continue } + if (sender === 'owner') { + const db = this.deps.adapter.openReadonly(scope.sessionId) + if (!db) { + failedSessionIds.add(scope.sessionId) + continue + } + const meta = getSessionMeta(db) + if (!meta?.ownerId?.trim()) { + if (ownerResolution) ownerResolution.missingOwnerSessions++ + continue + } + const owner = getMembers(db).find((member) => member.platformId === meta.ownerId) + if (!owner) { + if (ownerResolution) ownerResolution.unresolvedOwnerSessions++ + continue + } + if (ownerResolution) ownerResolution.resolvedSessions++ + candidates.push({ descriptor, memberIds: [owner.id] }) + continue + } candidates.push({ descriptor, memberIds: scope.memberIds }) } return { @@ -384,6 +491,16 @@ class DefaultCrossChatAnalysisService implements CrossChatAnalysisService { } } +function normalizeContactName(value: string): string { + return value.trim().toLocaleLowerCase() +} + +function normalizeRecentDays(value: number | undefined): number | undefined { + if (value === undefined) return undefined + if (!Number.isFinite(value) || value <= 0) throw new Error('recentDays must be a positive number') + return Math.min(MAX_RECENT_DAYS, Math.max(1, Math.floor(value))) +} + interface SearchCandidate { descriptor: CrossChatSessionDescriptor memberIds?: number[] @@ -450,6 +567,20 @@ function normalizeScopes(scopes: CrossChatSearchScope[]): CrossChatSearchScope[] return [...bySession.values()] } +function normalizeOverviewScopes(scopes: CrossChatSearchScope[]): CrossChatSearchScope[] { + return scopes.flatMap((scope) => { + const sessionId = scope.sessionId.trim() + if (!sessionId) return [] + return [ + { + sessionId, + memberIds: scope.memberIds ? [...new Set(scope.memberIds)] : undefined, + label: scope.label, + }, + ] + }) +} + function buildOverviewItem( descriptor: CrossChatSessionDescriptor, db: DatabaseAdapter, diff --git a/packages/node-runtime/src/services/cross-chat-analysis/types.ts b/packages/node-runtime/src/services/cross-chat-analysis/types.ts index 8b2e3fad..630a2e6e 100644 --- a/packages/node-runtime/src/services/cross-chat-analysis/types.ts +++ b/packages/node-runtime/src/services/cross-chat-analysis/types.ts @@ -1,140 +1,20 @@ -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'> - } -} +export type { + CrossChatEntityResolution, + CrossChatMessageContextRequest, + CrossChatMessageContextResult, + CrossChatMessageSource, + CrossChatOperationOptions, + CrossChatOverviewItem, + CrossChatOverviewRequest, + CrossChatOverviewResult, + CrossChatResolvedContact, + CrossChatResolvedContactSession, + CrossChatResolvedSession, + CrossChatSearchProgress, + CrossChatSearchRequest, + CrossChatSearchResult, + CrossChatSearchScope, + CrossChatSessionDescriptor, + CrossChatTruncationReason, + CrossChatUnresolvedEntity, +} from '@openchatlab/shared-types' diff --git a/packages/node-runtime/src/services/index.ts b/packages/node-runtime/src/services/index.ts index 298980b1..9e0c6e3e 100644 --- a/packages/node-runtime/src/services/index.ts +++ b/packages/node-runtime/src/services/index.ts @@ -54,7 +54,7 @@ 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 { createCrossChatAnalysisService, preprocessCrossChatMessages } from './cross-chat-analysis' export type { CrossChatAnalysisService, CrossChatAnalysisServiceDeps, diff --git a/packages/shared-types/index.ts b/packages/shared-types/index.ts index 0509ec1c..a7a587c9 100644 --- a/packages/shared-types/index.ts +++ b/packages/shared-types/index.ts @@ -123,6 +123,19 @@ export enum ChatType { PRIVATE = 'private', } +export type AIEntityRef = + | { + type: 'contact' + contactKey: string + displayName: string + } + | { + type: 'session' + sessionId: string + displayName: string + sessionType: 'private' | 'group' + } + export { CHATLAB_FORMAT_VERSION, CHATLAB_SUPPORTED_FORMAT_VERSIONS, @@ -497,6 +510,197 @@ export interface ContactDetailResponse { export type ContactsResponse = ContactsListResponse +// ==================== Cross-chat AI analysis ==================== + +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 CrossChatContactCandidate { + contactKey: string + displayName: string + platform: ChatPlatform + aliases: string[] + sourceSessions: Array<{ + id: string + name: string + type: ChatType + }> +} + +export interface CrossChatContactLookupResult { + query: string + status: 'resolved' | 'ambiguous' | 'not_found' | 'unavailable' + cacheStatus: ContactsCacheStatus + totalCandidates: number + candidates: CrossChatContactCandidate[] +} + +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 + recentDays?: number + sender?: 'all' | 'owner' + 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 + appliedFilters: { + startTs: number | null + endTs: number | null + recentDays: number | null + sender: 'all' | 'owner' + } + coverage: { + candidateSessions: number + scannedSessions: number + matchedSessions: number + failedSessions: number + ownerResolution?: { + resolvedSessions: number + missingOwnerSessions: number + unresolvedOwnerSessions: 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'> + } +} + +export interface CrossChatEvidenceSource { + sessionId: string + sessionName: string + sessionType: ChatType + platform: ChatPlatform + messageId: number + senderName: string + timestamp: number + snippet: string +} + +export interface CrossChatEvidencePayload { + version: 1 + query: string + sources: CrossChatEvidenceSource[] + coverage: CrossChatSearchResult['coverage'] +} + // ==================== Global Insight ==================== export type AnnualSummaryMode = 'year' | 'recent' diff --git a/packages/tools/src/agent-adapter.ts b/packages/tools/src/agent-adapter.ts index 11b249fb..c2cc7ee9 100644 --- a/packages/tools/src/agent-adapter.ts +++ b/packages/tools/src/agent-adapter.ts @@ -1,4 +1,4 @@ -import type { JsonSchema, ToolDefinition, ToolExecutionContext, ToolResult } from './types' +import type { JsonSchema, ToolDefinition, ToolResult } from './types' interface AgentToolParameters { type: 'object' @@ -51,10 +51,10 @@ function toAgentToolResult(result: ToolResult): AgentToolExecutionResult { } /** 执行共享工具并把异常收敛成 Agent 可直接返回给模型的文本结果。 */ -export async function executeToolForAgent( - tool: ToolDefinition, +export async function executeToolForAgent( + tool: ToolDefinition, params: unknown, - context: ToolExecutionContext + context: TContext ): Promise { const toolParams = (params && typeof params === 'object' ? params : {}) as Record try { diff --git a/packages/tools/src/definitions/cross-chat-tools.test.ts b/packages/tools/src/definitions/cross-chat-tools.test.ts new file mode 100644 index 00000000..d08d545f --- /dev/null +++ b/packages/tools/src/definitions/cross-chat-tools.test.ts @@ -0,0 +1,472 @@ +import assert from 'node:assert/strict' +import { describe, it } from 'node:test' +import type { + AIEntityRef, + CrossChatContactLookupResult, + CrossChatEntityResolution, + CrossChatMessageContextResult, + CrossChatMessageSource, + CrossChatOverviewResult, + CrossChatSearchResult, +} from '@openchatlab/shared-types' +import { ChatType } from '@openchatlab/shared-types' +import { AGENT_TOOL_REGISTRY, CROSS_CHAT_AGENT_TOOL_REGISTRY, MCP_TOOL_REGISTRY } from '../registry' +import type { CrossChatAnalysisToolService, CrossChatToolExecutionContext } from '../types' + +function messageSource( + sessionId: string, + messageId: number, + timestamp: number, + content = `message-${messageId}` +): CrossChatMessageSource { + return { + sessionId, + sessionName: `Session ${sessionId}`, + sessionType: ChatType.PRIVATE, + platform: 'test', + lastMessageTs: timestamp, + messageId, + senderId: 2, + senderName: 'Alice', + senderPlatformId: 'alice', + content, + timestamp, + messageType: 0, + } +} + +function createContext(overrides: Partial = {}): CrossChatToolExecutionContext { + const service: CrossChatAnalysisToolService = { + lookupContact: (query: string): CrossChatContactLookupResult => ({ + query, + status: 'not_found', + cacheStatus: 'fresh', + totalCandidates: 0, + candidates: [], + }), + resolveEntities: (_refs: AIEntityRef[]): CrossChatEntityResolution => ({ + contacts: [], + sessions: [], + unresolved: [], + coverage: { + requestedEntities: 0, + resolvedEntities: 0, + candidateSessions: 0, + resolvedSessions: 0, + failedSessions: 0, + }, + }), + searchMessages: async (): Promise => ({ + messages: [ + { + sessionId: 'session-a', + sessionName: 'Session A', + sessionType: ChatType.PRIVATE, + platform: 'test', + lastMessageTs: 10, + messageId: 7, + senderId: 2, + senderName: 'Alice', + senderPlatformId: 'alice', + content: 'private raw content', + timestamp: 10, + messageType: 0, + }, + ], + totalMatches: 1, + appliedFilters: { + startTs: null, + endTs: null, + recentDays: null, + sender: 'all', + }, + coverage: { + candidateSessions: 1, + scannedSessions: 1, + matchedSessions: 1, + failedSessions: 0, + truncated: false, + truncatedReasons: [], + }, + }), + getMessageContext: (): CrossChatMessageContextResult => ({ + source: { + sessionId: 'session-a', + sessionName: 'Session A', + sessionType: ChatType.PRIVATE, + platform: 'test', + lastMessageTs: 10, + }, + messages: [], + }), + getOverview: async (): Promise => ({ + items: [], + coverage: { + candidateSessions: 0, + analyzedSessions: 0, + failedSessions: 0, + truncated: false, + truncatedReasons: [], + }, + }), + ...overrides, + } + return { + locale: 'zh-CN', + analysisService: service, + preprocessMessagesBySession: (_sessionId, messages) => + messages.map((message) => ({ ...message, senderName: 'U1', content: '[redacted]' })), + } +} + +describe('cross-chat agent registry', () => { + it('contains only the four dedicated tools and is isolated from session and MCP registries', () => { + const names = CROSS_CHAT_AGENT_TOOL_REGISTRY.map((tool) => tool.name) + assert.deepEqual(names, [ + 'resolve_chat_entities', + 'search_messages_globally', + 'get_cross_chat_message_context', + 'get_cross_chat_overview', + ]) + for (const name of names) { + assert.equal( + AGENT_TOOL_REGISTRY.some((tool) => tool.name === name), + false + ) + assert.equal( + MCP_TOOL_REGISTRY.some((tool) => tool.name === name), + false + ) + } + }) + + it('resolves a unique contact name before continuing with stable scopes', async () => { + let resolvedRefs: AIEntityRef[] = [] + const context = createContext({ + lookupContact: () => ({ + query: '小红', + status: 'resolved', + cacheStatus: 'fresh', + totalCandidates: 1, + candidates: [ + { + contactKey: 'test:xiaohong', + displayName: '小红', + platform: 'test', + aliases: [], + sourceSessions: [{ id: 'private-xiaohong', name: '小红', type: ChatType.PRIVATE }], + }, + ], + }), + resolveEntities: (refs) => { + resolvedRefs = refs + return { + contacts: [], + sessions: [], + unresolved: [], + coverage: { + requestedEntities: refs.length, + resolvedEntities: refs.length, + candidateSessions: 1, + resolvedSessions: 1, + failedSessions: 0, + }, + } + }, + } as Partial) + const tool = CROSS_CHAT_AGENT_TOOL_REGISTRY.find((item) => item.name === 'resolve_chat_entities') + assert.ok(tool) + + const result = await tool.handler({ entities: [{ type: 'contact', displayName: '小红' }] }, context) + assert.deepEqual(resolvedRefs, [{ type: 'contact', contactKey: 'test:xiaohong', displayName: '小红' }]) + assert.equal((result.data as { contactLookups: Array<{ status: string }> }).contactLookups[0]?.status, 'resolved') + }) + + it('returns ambiguous contact candidates without choosing one', async () => { + let resolvedRefs: AIEntityRef[] = [] + const context = createContext({ + lookupContact: () => ({ + query: '小红', + status: 'ambiguous', + cacheStatus: 'fresh', + totalCandidates: 2, + candidates: [ + { + contactKey: 'test:xiaohong-1', + displayName: '小红', + platform: 'test', + aliases: ['小红 A'], + sourceSessions: [{ id: 'private-1', name: '小红 A', type: ChatType.PRIVATE }], + }, + { + contactKey: 'test:xiaohong-2', + displayName: '小红', + platform: 'test', + aliases: ['小红 B'], + sourceSessions: [{ id: 'private-2', name: '小红 B', type: ChatType.PRIVATE }], + }, + ], + }), + resolveEntities: (refs) => { + resolvedRefs = refs + return { + contacts: [], + sessions: [], + unresolved: [], + coverage: { + requestedEntities: refs.length, + resolvedEntities: 0, + candidateSessions: 0, + resolvedSessions: 0, + failedSessions: 0, + }, + } + }, + } as Partial) + const tool = CROSS_CHAT_AGENT_TOOL_REGISTRY.find((item) => item.name === 'resolve_chat_entities') + assert.ok(tool) + + const result = await tool.handler({ entities: [{ type: 'contact', displayName: '小红' }] }, context) + assert.deepEqual(resolvedRefs, []) + assert.equal((result.data as { contactLookups: Array<{ status: string }> }).contactLookups[0]?.status, 'ambiguous') + }) + + it('sanitizes each session before returning global search evidence', async () => { + const tool = CROSS_CHAT_AGENT_TOOL_REGISTRY.find((item) => item.name === 'search_messages_globally') + assert.ok(tool) + const result = await tool.handler({ keywords: ['private'] }, createContext()) + const data = result.data as { + crossChatEvidence: { sources: Array<{ sessionId: string; messageId: number; snippet: string }> } + } + + assert.equal(result.content.includes('private raw content'), false) + assert.equal(result.content.includes('[redacted]'), true) + assert.equal(result.content.includes('crossChatEvidence'), false) + assert.deepEqual(data.crossChatEvidence.sources, [ + { + sessionId: 'session-a', + sessionName: 'Session A', + sessionType: 'private', + platform: 'test', + messageId: 7, + senderName: 'U1', + timestamp: 10, + snippet: '[redacted]', + }, + ]) + }) + + it('preserves the requested search ordering across per-session preprocessing', async () => { + const context = createContext({ + searchMessages: async () => ({ + messages: [messageSource('a', 1, 1), messageSource('b', 1, 2), messageSource('a', 2, 3)], + totalMatches: 3, + appliedFilters: { startTs: null, endTs: null, recentDays: null, sender: 'all' }, + coverage: { + candidateSessions: 2, + scannedSessions: 2, + matchedSessions: 2, + failedSessions: 0, + truncated: false, + truncatedReasons: [], + }, + }), + }) + const tool = CROSS_CHAT_AGENT_TOOL_REGISTRY.find((item) => item.name === 'search_messages_globally') + assert.ok(tool) + + const result = await tool.handler({ keywords: ['message'], sort: 'asc' }, context) + const messages = (result.data as { messages: Array<{ sessionId: string; messageId: number }> }).messages + + assert.deepEqual( + messages.map((message) => [message.sessionId, message.messageId]), + [ + ['a', 1], + ['b', 1], + ['a', 2], + ] + ) + }) + + it('budgets the complete model-visible search payload without duplicating evidence snippets', async () => { + const context = createContext({ + searchMessages: async () => ({ + messages: [messageSource('a', 1, 1, 'x'.repeat(500))], + totalMatches: 1, + appliedFilters: { startTs: null, endTs: null, recentDays: null, sender: 'all' }, + coverage: { + candidateSessions: 1, + scannedSessions: 1, + matchedSessions: 1, + failedSessions: 0, + truncated: false, + truncatedReasons: [], + }, + }), + }) + context.maxToolResultTokens = 300 + context.preprocessMessagesBySession = (_sessionId, messages) => messages + const tool = CROSS_CHAT_AGENT_TOOL_REGISTRY.find((item) => item.name === 'search_messages_globally') + assert.ok(tool) + + const result = await tool.handler({ keywords: ['x'] }, context) + const modelData = JSON.parse(result.content) as Record + const details = result.data as { + returned: number + crossChatEvidence: { sources: Array<{ snippet: string }> } + } + + assert.ok(result.content.length <= 300 * 4) + assert.equal('crossChatEvidence' in modelData, false) + assert.equal(details.returned, 1) + assert.equal(details.crossChatEvidence.sources[0]?.snippet, 'x'.repeat(500)) + }) + + it('allows recent-message sampling only when explicit scopes are present', async () => { + let captured: unknown + const context = createContext({ + searchMessages: async (request) => { + captured = request + return { + messages: [], + totalMatches: 0, + appliedFilters: { + startTs: null, + endTs: null, + recentDays: null, + sender: 'all', + }, + coverage: { + candidateSessions: 1, + scannedSessions: 1, + matchedSessions: 0, + failedSessions: 0, + truncated: false, + truncatedReasons: [], + }, + } + }, + }) + const tool = CROSS_CHAT_AGENT_TOOL_REGISTRY.find((item) => item.name === 'search_messages_globally') + assert.ok(tool) + await tool.handler({ scopes: [{ sessionId: 'session-a', memberIds: [2] }] }, context) + assert.deepEqual(captured, { + keywords: [], + scopes: [{ sessionId: 'session-a', memberIds: [2], label: undefined }], + startTs: undefined, + endTs: undefined, + recentDays: undefined, + sender: 'all', + matchMode: 'any', + sort: 'desc', + maxSessions: undefined, + maxEvidence: undefined, + maxWallTimeMs: undefined, + }) + }) + + it('forwards relative time and owner-only filters for global discovery', async () => { + let captured: unknown + const context = createContext({ + searchMessages: async (request) => { + captured = request + return { + messages: [], + totalMatches: 0, + appliedFilters: { + startTs: 100, + endTs: null, + recentDays: 90, + sender: 'owner', + }, + coverage: { + candidateSessions: 1, + scannedSessions: 1, + matchedSessions: 0, + failedSessions: 0, + ownerResolution: { + resolvedSessions: 1, + missingOwnerSessions: 0, + unresolvedOwnerSessions: 0, + }, + truncated: false, + truncatedReasons: [], + }, + } + }, + }) + const tool = CROSS_CHAT_AGENT_TOOL_REGISTRY.find((item) => item.name === 'search_messages_globally') + assert.ok(tool) + await tool.handler({ keywords: ['买房'], recent_days: 90, sender: 'owner' }, context) + + assert.deepEqual(captured, { + keywords: ['买房'], + scopes: undefined, + startTs: undefined, + endTs: undefined, + recentDays: 90, + sender: 'owner', + matchMode: 'any', + sort: 'desc', + maxSessions: undefined, + maxEvidence: undefined, + maxWallTimeMs: undefined, + }) + }) + + it('requires compound source identity for cross-chat context lookup', async () => { + let captured: unknown + const context = createContext({ + getMessageContext: (request) => { + captured = request + return { + source: { + sessionId: request.sessionId, + sessionName: 'Session A', + sessionType: ChatType.PRIVATE, + platform: 'test', + lastMessageTs: null, + }, + messages: [], + } + }, + }) + const tool = CROSS_CHAT_AGENT_TOOL_REGISTRY.find((item) => item.name === 'get_cross_chat_message_context') + assert.ok(tool) + await tool.handler({ session_id: 'session-a', message_id: 7, context_size: 3 }, context) + assert.deepEqual(captured, { sessionId: 'session-a', messageId: 7, contextSize: 3 }) + }) + + it('retains the requested context anchor under a small model result budget', async () => { + const context = createContext({ + getMessageContext: () => ({ + source: { + sessionId: 'session-a', + sessionName: 'Session A', + sessionType: ChatType.PRIVATE, + platform: 'test', + lastMessageTs: 101, + }, + messages: Array.from({ length: 101 }, (_, index) => + messageSource('session-a', index + 1, index + 1, 'x'.repeat(500)) + ), + }), + }) + context.maxToolResultTokens = 1024 + context.preprocessMessagesBySession = (_sessionId, messages) => messages + const tool = CROSS_CHAT_AGENT_TOOL_REGISTRY.find((item) => item.name === 'get_cross_chat_message_context') + assert.ok(tool) + + const result = await tool.handler({ session_id: 'session-a', message_id: 51, context_size: 50 }, context) + const data = result.data as { truncated: boolean; messages: Array<{ messageId: number }> } + const ids = data.messages.map((message) => message.messageId) + + assert.equal(data.truncated, true) + assert.equal(ids.includes(51), true) + assert.deepEqual( + ids, + [...ids].sort((left, right) => left - right) + ) + assert.ok(result.content.length <= 1024 * 4) + }) +}) diff --git a/packages/tools/src/definitions/cross-chat-tools.ts b/packages/tools/src/definitions/cross-chat-tools.ts new file mode 100644 index 00000000..5296e1f8 --- /dev/null +++ b/packages/tools/src/definitions/cross-chat-tools.ts @@ -0,0 +1,452 @@ +import type { + AIEntityRef, + CrossChatContactLookupResult, + CrossChatEvidencePayload, + CrossChatMessageSource, + CrossChatSearchScope, +} from '@openchatlab/shared-types' +import type { CrossChatToolExecutionContext, JsonSchema, ToolDefinition, ToolResult } from '../types' +import { parseExtendedTimeParams } from '../utils/time-params' +import { timeParamProperties } from '../utils/schemas' + +const scopeItems = { + type: 'object', + properties: { + sessionId: { type: 'string', description: 'Stable session ID returned by resolve_chat_entities' }, + memberIds: { type: 'array', items: { type: 'number' }, description: 'Local member IDs for this session' }, + label: { type: 'string', description: 'Human-readable scope label' }, + }, + required: ['sessionId'], +} + +const resolveSchema: JsonSchema = { + type: 'object', + properties: { + entities: { + type: 'array', + description: + 'Contact or session references. A contact may use a stable contactKey or a displayName lookup from the user text.', + items: { + type: 'object', + properties: { + type: { type: 'string', enum: ['contact', 'session'] }, + contactKey: { type: 'string' }, + sessionId: { type: 'string' }, + displayName: { type: 'string' }, + sessionType: { type: 'string', enum: ['private', 'group'] }, + }, + required: ['type', 'displayName'], + }, + }, + }, + required: ['entities'], +} + +const searchSchema: JsonSchema = { + type: 'object', + properties: { + keywords: { + type: 'array', + items: { type: 'string' }, + description: + 'Exact substring keywords. May be omitted only with explicit scopes to sample recent messages from those scopes.', + }, + scopes: { + type: 'array', + description: 'Resolved session/member scopes. Omit only when the user clearly requested global discovery.', + items: scopeItems, + }, + match_mode: { type: 'string', enum: ['any', 'all'], description: 'Whether any or all keywords must match' }, + recent_days: { + type: 'number', + description: + 'Relative time window ending now. Use 90 when the user says recent/recently without a more specific range.', + }, + sender: { + type: 'string', + enum: ['all', 'owner'], + description: + "Message sender filter. Use owner when the user's wording makes them the subject, such as 'I discussed buying a home'.", + }, + sort: { type: 'string', enum: ['asc', 'desc'], description: 'Timestamp order; defaults to newest first' }, + max_sessions: { type: 'number', description: 'Maximum sessions to scan' }, + max_evidence: { type: 'number', description: 'Maximum evidence messages returned' }, + max_wall_time_ms: { type: 'number', description: 'Maximum wall time for this scan' }, + ...timeParamProperties, + }, +} + +const contextSchema: JsonSchema = { + type: 'object', + properties: { + session_id: { type: 'string', description: 'Source session ID from cross-chat search evidence' }, + message_id: { type: 'number', description: 'Message ID inside that source session' }, + context_size: { type: 'number', description: 'Messages before and after the source message; defaults to 10' }, + }, + required: ['session_id', 'message_id'], +} + +const overviewSchema: JsonSchema = { + type: 'object', + properties: { + scopes: { + type: 'array', + description: 'Resolved session/member scopes to summarize separately', + items: scopeItems, + }, + max_sessions: { type: 'number', description: 'Maximum sessions to analyze' }, + max_wall_time_ms: { type: 'number', description: 'Maximum wall time for this analysis' }, + }, + required: ['scopes'], +} + +function resolveHandler(params: Record, context: CrossChatToolExecutionContext): ToolResult { + const inputs = parseEntityInputs(params.entities) + const refs: AIEntityRef[] = [] + const contactLookups: CrossChatContactLookupResult[] = [] + for (const input of inputs) { + if (input.type === 'contact' && !input.contactKey) { + const lookup = context.analysisService.lookupContact(input.displayName) + contactLookups.push(lookup) + if (lookup.status === 'resolved' && lookup.candidates[0]) { + refs.push({ + type: 'contact', + contactKey: lookup.candidates[0].contactKey, + displayName: lookup.candidates[0].displayName, + }) + } + continue + } + refs.push(input as AIEntityRef) + } + const resolution = context.analysisService.resolveEntities(dedupeEntityRefs(refs)) + const data = { ...resolution, contactLookups } + return { content: JSON.stringify(data), data } +} + +async function searchHandler( + params: Record, + context: CrossChatToolExecutionContext +): Promise { + const timeFilter = parseExtendedTimeParams(params) + const keywords = params.keywords === undefined ? [] : parseStringArray(params.keywords) + const result = await context.analysisService.searchMessages( + { + keywords, + scopes: params.scopes === undefined ? undefined : parseScopes(params.scopes), + startTs: timeFilter?.startTs, + endTs: timeFilter?.endTs, + recentDays: parseOptionalNumber(params.recent_days), + sender: params.sender === 'owner' ? 'owner' : 'all', + matchMode: params.match_mode === 'all' ? 'all' : 'any', + sort: params.sort === 'asc' ? 'asc' : 'desc', + maxSessions: parseOptionalNumber(params.max_sessions), + maxEvidence: parseOptionalNumber(params.max_evidence), + maxWallTimeMs: parseOptionalNumber(params.max_wall_time_ms), + }, + { + signal: context.abortSignal, + onProgress: (progress) => + context.reportProgress?.({ + phase: 'searching', + current: progress.processedSessions, + total: progress.totalSessions, + }), + } + ) + const safeMessages = await preprocessBySession(context, result.messages) + const buildCoverage = (budgetTruncated: boolean) => ({ + ...result.coverage, + truncated: result.coverage.truncated || budgetTruncated, + truncatedReasons: budgetTruncated + ? [...new Set([...result.coverage.truncatedReasons, 'evidence_budget' as const])] + : result.coverage.truncatedReasons, + }) + const buildModelData = (messages: CrossChatMessageSource[], budgetTruncated: boolean) => ({ + totalMatches: result.totalMatches, + returned: messages.length, + appliedFilters: result.appliedFilters, + coverage: buildCoverage(budgetTruncated), + messages: messages.map(toModelMessage), + }) + const limited = limitMessagesToBudget(safeMessages, context.maxToolResultTokens, buildModelData) + const modelData = buildModelData(limited.messages, limited.truncated) + const evidence: CrossChatEvidencePayload = { + version: 1, + query: keywords.join(' '), + sources: limited.messages.map(toEvidenceSource), + coverage: modelData.coverage, + } + const data = { + ...modelData, + crossChatEvidence: evidence, + } + // Evidence details are persisted for source cards, but duplicating their snippets in content would charge the + // model twice for the same text and invalidate maxToolResultTokens. + return { content: JSON.stringify(modelData), data } +} + +async function contextHandler( + params: Record, + context: CrossChatToolExecutionContext +): Promise { + const sessionId = requireString(params.session_id, 'session_id') + const messageId = requireNumber(params.message_id, 'message_id') + const contextSize = parseOptionalNumber(params.context_size) + const result = context.analysisService.getMessageContext({ sessionId, messageId, contextSize }) + const safeMessages = await preprocessBySession(context, result.messages) + const buildModelData = (messages: CrossChatMessageSource[], truncated: boolean) => ({ + source: result.source, + returned: messages.length, + truncated, + messages: messages.map(toModelMessage), + }) + const limited = limitMessagesToBudget(safeMessages, context.maxToolResultTokens, buildModelData, { + priorityIndexes: buildContextPriority(safeMessages, messageId), + continueAfterOverflow: true, + }) + const data = buildModelData(limited.messages, limited.truncated) + return { content: JSON.stringify(data), data } +} + +async function overviewHandler( + params: Record, + context: CrossChatToolExecutionContext +): Promise { + const result = await context.analysisService.getOverview( + { + scopes: parseScopes(params.scopes), + maxSessions: parseOptionalNumber(params.max_sessions), + maxWallTimeMs: parseOptionalNumber(params.max_wall_time_ms), + }, + { + signal: context.abortSignal, + onProgress: (progress) => + context.reportProgress?.({ + phase: 'analyzing', + current: progress.processedSessions, + total: progress.totalSessions, + }), + } + ) + const data = { + items: result.items.map(({ memberNames: _memberNames, ...item }) => item), + coverage: result.coverage, + } + return { content: JSON.stringify(data), data } +} + +export const resolveChatEntitiesTool: ToolDefinition = { + name: 'resolve_chat_entities', + description: + 'Resolve contacts and sessions into exact source scopes. Stable refs resolve directly; typed contact names search the contact catalog. Continue automatically for one candidate, ask the user to choose when contactLookups reports ambiguous, and never guess among multiple candidates.', + inputSchema: resolveSchema, + handler: resolveHandler, + category: 'core', +} + +export const searchMessagesGloballyTool: ToolDefinition = { + name: 'search_messages_globally', + description: + "Search exact keywords across resolved contacts or sessions. Use recent_days=90 when 'recent' has no explicit duration, and sender=owner when the user is the subject of the requested action. Owner hits are discovery seeds; expand their context to identify other participants. With explicit scopes, omit keywords to sample recent messages. Unscoped global discovery always requires keywords. Results include applied filters, source identity, coverage, and truncation status.", + inputSchema: searchSchema, + handler: searchHandler, + category: 'core', + truncationStrategy: 'keep_first', +} + +export const getCrossChatMessageContextTool: ToolDefinition = { + name: 'get_cross_chat_message_context', + description: + 'Load surrounding messages for one cross-chat evidence item. Both session_id and message_id are required because message IDs are only unique inside a session.', + inputSchema: contextSchema, + handler: contextHandler, + category: 'core', + truncationStrategy: 'keep_last', +} + +export const getCrossChatOverviewTool: ToolDefinition = { + name: 'get_cross_chat_overview', + description: + 'Get separate message-count and time-range overviews for resolved contact/session scopes. This is a basic comparison tool, not arbitrary SQL or single-chat deep analytics.', + inputSchema: overviewSchema, + handler: overviewHandler, + category: 'core', +} + +type ParsedEntityInput = + | { type: 'contact'; contactKey?: string; displayName: string } + | Extract + +function parseEntityInputs(value: unknown): ParsedEntityInput[] { + if (!Array.isArray(value)) throw new Error('entities must be an array') + return value.map((item) => { + if (!isRecord(item)) throw new Error('each entity must be an object') + const type = requireString(item.type, 'entity.type') + const displayName = requireString(item.displayName, 'entity.displayName') + if (type === 'contact') { + return { + type, + ...(typeof item.contactKey === 'string' && item.contactKey.trim() + ? { contactKey: item.contactKey.trim() } + : {}), + displayName, + } + } + if (type === 'session') { + const sessionType = item.sessionType === 'private' ? 'private' : item.sessionType === 'group' ? 'group' : null + if (!sessionType) throw new Error('entity.sessionType must be private or group') + return { type, sessionId: requireString(item.sessionId, 'entity.sessionId'), displayName, sessionType } + } + throw new Error('entity.type must be contact or session') + }) +} + +function dedupeEntityRefs(refs: AIEntityRef[]): AIEntityRef[] { + const byKey = new Map() + for (const ref of refs) { + const key = ref.type === 'contact' ? `contact:${ref.contactKey}` : `session:${ref.sessionId}` + byKey.set(key, ref) + } + return [...byKey.values()] +} + +function parseScopes(value: unknown): CrossChatSearchScope[] { + if (!Array.isArray(value)) throw new Error('scopes must be an array') + return value.map((item) => { + if (!isRecord(item)) throw new Error('each scope must be an object') + return { + sessionId: requireString(item.sessionId, 'scope.sessionId'), + memberIds: item.memberIds === undefined ? undefined : parseNumberArray(item.memberIds), + label: typeof item.label === 'string' ? item.label : undefined, + } + }) +} + +async function preprocessBySession( + context: CrossChatToolExecutionContext, + messages: CrossChatMessageSource[] +): Promise { + const bySession = new Map() + for (const message of messages) { + const group = bySession.get(message.sessionId) ?? [] + group.push({ ...message }) + bySession.set(message.sessionId, group) + } + const safe: CrossChatMessageSource[] = [] + const processedBySession = new Map>() + for (const [sessionId, group] of bySession) { + const processed = await context.preprocessMessagesBySession(sessionId, group) + const byMessageId = new Map() + for (const message of processed) { + if (message.sessionId === sessionId) byMessageId.set(message.messageId, message) + } + processedBySession.set(sessionId, byMessageId) + } + for (const message of messages) { + const processed = processedBySession.get(message.sessionId)?.get(message.messageId) + if (processed) safe.push(processed) + } + return safe +} + +function limitMessagesToBudget( + messages: CrossChatMessageSource[], + maxToolResultTokens: number | undefined, + buildPayload: (messages: CrossChatMessageSource[], truncated: boolean) => unknown, + options: { priorityIndexes?: number[]; continueAfterOverflow?: boolean } = {} +): { messages: CrossChatMessageSource[]; truncated: boolean } { + const prepared = messages.map((message) => ({ + ...message, + content: message.content.length > 500 ? `${message.content.slice(0, 500)}…` : message.content, + })) + if (!maxToolResultTokens || maxToolResultTokens <= 0) { + return { messages: prepared, truncated: false } + } + + const maxChars = maxToolResultTokens * 4 + const priorityIndexes = options.priorityIndexes ?? prepared.map((_, index) => index) + const selectedIndexes = new Set() + for (const index of priorityIndexes) { + if (index < 0 || index >= prepared.length || selectedIndexes.has(index)) continue + const candidateIndexes = [...selectedIndexes, index].sort((left, right) => left - right) + const candidateMessages = candidateIndexes.map((candidateIndex) => prepared[candidateIndex]) + const truncated = candidateMessages.length < prepared.length + const estimatedChars = JSON.stringify(buildPayload(candidateMessages, truncated)).length + if (estimatedChars > maxChars) { + if (options.continueAfterOverflow) continue + break + } + selectedIndexes.add(index) + } + const limited = [...selectedIndexes].sort((left, right) => left - right).map((index) => prepared[index]) + return { messages: limited, truncated: limited.length < prepared.length } +} + +function buildContextPriority(messages: CrossChatMessageSource[], anchorMessageId: number): number[] { + const anchorIndex = messages.findIndex((message) => message.messageId === anchorMessageId) + if (anchorIndex < 0) return messages.map((_, index) => index) + + const indexes = [anchorIndex] + for (let distance = 1; indexes.length < messages.length; distance++) { + const before = anchorIndex - distance + const after = anchorIndex + distance + if (before >= 0) indexes.push(before) + if (after < messages.length) indexes.push(after) + } + return indexes +} + +function toModelMessage(message: CrossChatMessageSource): Record { + return { + sessionId: message.sessionId, + sessionName: message.sessionName, + sessionType: message.sessionType, + messageId: message.messageId, + senderId: message.senderId, + senderName: message.senderName, + timestamp: message.timestamp, + content: message.content, + } +} + +function toEvidenceSource(message: CrossChatMessageSource): CrossChatEvidencePayload['sources'][number] { + return { + sessionId: message.sessionId, + sessionName: message.sessionName, + sessionType: message.sessionType, + platform: message.platform, + messageId: message.messageId, + senderName: message.senderName, + timestamp: message.timestamp, + snippet: message.content, + } +} + +function parseStringArray(value: unknown): string[] { + if (!Array.isArray(value)) throw new Error('keywords must be an array') + return value.map((item) => requireString(item, 'keyword')).filter(Boolean) +} + +function parseNumberArray(value: unknown): number[] { + if (!Array.isArray(value)) throw new Error('memberIds must be an array') + return value.map((item) => requireNumber(item, 'memberId')) +} + +function requireString(value: unknown, name: string): string { + if (typeof value !== 'string' || !value.trim()) throw new Error(`${name} is required`) + return value.trim() +} + +function requireNumber(value: unknown, name: string): number { + if (typeof value !== 'number' || !Number.isFinite(value)) throw new Error(`${name} must be a number`) + return value +} + +function parseOptionalNumber(value: unknown): number | undefined { + return typeof value === 'number' && Number.isFinite(value) ? value : undefined +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value) +} diff --git a/packages/tools/src/index.ts b/packages/tools/src/index.ts index 0c518fd0..3c88e7cc 100644 --- a/packages/tools/src/index.ts +++ b/packages/tools/src/index.ts @@ -9,6 +9,7 @@ export { MCP_TOOL_REGISTRY, AGENT_TOOL_REGISTRY, + CROSS_CHAT_AGENT_TOOL_REGISTRY, SEMANTIC_SEARCH_TOOL_NAME, RETRIEVE_CHAT_EVIDENCE_TOOL_NAME, getToolByName, @@ -41,6 +42,12 @@ export { keywordFrequencyTool } from './definitions/keyword-frequency' export { renderChartTool } from './definitions/render-chart' export { semanticSearchCurrentChatTool } from './definitions/semantic-search-current-chat' export { retrieveChatEvidenceTool } from './definitions/retrieve-chat-evidence' +export { + getCrossChatMessageContextTool, + getCrossChatOverviewTool, + resolveChatEntitiesTool, + searchMessagesGloballyTool, +} from './definitions/cross-chat-tools' // === SQL Tools === export { SQL_TOOL_DEFS, createSqlToolDefinition, createAllSqlToolDefinitions } from './sql' @@ -53,6 +60,8 @@ export { parseExtendedTimeParams } from './utils/time-params' export type { ToolDefinition, ToolExecutionContext, + CrossChatAnalysisToolService, + CrossChatToolExecutionContext, ToolProgress, ToolResult, JsonSchema, diff --git a/packages/tools/src/registry.ts b/packages/tools/src/registry.ts index 53197647..f38f4cbf 100644 --- a/packages/tools/src/registry.ts +++ b/packages/tools/src/registry.ts @@ -30,6 +30,12 @@ import { responseTimeAnalysisTool } from './definitions/response-time-analysis' import { keywordFrequencyTool } from './definitions/keyword-frequency' import { semanticSearchCurrentChatTool } from './definitions/semantic-search-current-chat' import { retrieveChatEvidenceTool } from './definitions/retrieve-chat-evidence' +import { + getCrossChatMessageContextTool, + getCrossChatOverviewTool, + resolveChatEntitiesTool, + searchMessagesGloballyTool, +} from './definitions/cross-chat-tools' import { SQL_TOOL_DEFS, createAllSqlToolDefinitions } from './sql' /** @@ -75,6 +81,14 @@ export const AGENT_TOOL_REGISTRY: ToolDefinition[] = [ ...createAllSqlToolDefinitions(SQL_TOOL_DEFS), ] +/** Dedicated global-analysis tools. Never merge this registry into session Agent or MCP registries. */ +export const CROSS_CHAT_AGENT_TOOL_REGISTRY = [ + resolveChatEntitiesTool, + searchMessagesGloballyTool, + getCrossChatMessageContextTool, + getCrossChatOverviewTool, +] + /** 语义检索工具名（runner 动态过滤用） */ export const SEMANTIC_SEARCH_TOOL_NAME = semanticSearchCurrentChatTool.name diff --git a/packages/tools/src/types.ts b/packages/tools/src/types.ts index 65d448c2..e8a80acc 100644 --- a/packages/tools/src/types.ts +++ b/packages/tools/src/types.ts @@ -5,7 +5,20 @@ */ import type { ChartPayload, DatabaseAdapter, EvidenceTimeRangeMs } from '@openchatlab/core' -import type { ToolProgress } from '@openchatlab/shared-types' +import type { + AIEntityRef, + CrossChatContactLookupResult, + CrossChatEntityResolution, + CrossChatMessageContextRequest, + CrossChatMessageContextResult, + CrossChatMessageSource, + CrossChatOperationOptions, + CrossChatOverviewRequest, + CrossChatOverviewResult, + CrossChatSearchRequest, + CrossChatSearchResult, + ToolProgress, +} from '@openchatlab/shared-types' export type { ToolProgress } from '@openchatlab/shared-types' @@ -32,7 +45,11 @@ export interface JsonSchema { { type: string description?: string - items?: { type: string } + items?: { + type: string + properties?: Record + required?: string[] + } properties?: Record additionalProperties?: boolean | Record default?: unknown @@ -338,6 +355,26 @@ export interface ToolExecutionContext { desensitizeMessages?: (messages: RawMessage[], options?: { anonymizeNames?: boolean }) => RawMessage[] } +export interface CrossChatAnalysisToolService { + lookupContact(query: string): CrossChatContactLookupResult + resolveEntities(refs: AIEntityRef[]): CrossChatEntityResolution + searchMessages(request: CrossChatSearchRequest, options?: CrossChatOperationOptions): Promise + getMessageContext(request: CrossChatMessageContextRequest): CrossChatMessageContextResult + getOverview(request: CrossChatOverviewRequest, options?: CrossChatOperationOptions): Promise +} + +export interface CrossChatToolExecutionContext { + locale?: string + abortSignal?: AbortSignal + reportProgress?: (progress: ToolProgress) => void + maxToolResultTokens?: number + analysisService: CrossChatAnalysisToolService + preprocessMessagesBySession: ( + sessionId: string, + messages: CrossChatMessageSource[] + ) => CrossChatMessageSource[] | Promise +} + // ==================== Raw Message ==================== /** @@ -377,11 +414,11 @@ export type ToolExecutionMode = 'parallel' | 'sequential' * * handler 支持同步和异步返回，兼容 Server（同步 DB）和 Electron（异步 Worker）。 */ -export interface ToolDefinition { +export interface ToolDefinition { name: string description: string inputSchema: JsonSchema - handler: (params: Record, context: ToolExecutionContext) => ToolResult | Promise + handler: (params: Record, context: TContext) => ToolResult | Promise category?: ToolCategory truncationStrategy?: TruncationStrategy /** Override the Agent's parallel default when this tool has an ordering dependency. */ diff --git a/src/services/ai-stream/types.ts b/src/services/ai-stream/types.ts index 0a35595f..e3eb6ae3 100644 --- a/src/services/ai-stream/types.ts +++ b/src/services/ai-stream/types.ts @@ -1,5 +1,5 @@ import type { PlanContentBlock } from '@/services/ai/planBlocks' -import type { ChartAutoMode, ToolProgress } from '@openchatlab/shared-types' +import type { AIEntityRef, ChartAutoMode, ToolProgress } from '@openchatlab/shared-types' export interface LlmStreamChunk { content: string @@ -106,7 +106,9 @@ export interface AgentStreamResult { export interface AgentStreamParams { userMessage: string - sessionId: string + chatKind?: 'session' | 'global' + sessionId?: string + entityRefs?: AIEntityRef[] aiChatId?: string historyLeafMessageId?: string | null chatType?: 'group' | 'private'