refactor(node-runtime): extract event-handler to @openchatlab/node-runtime

Move AgentEventHandler, TokenUsage, AgentRuntimeStatus, AgentStreamChunk
to packages/node-runtime/src/ai/agent/event-handler.ts.
Token estimation and event-to-chunk mapping are now shared.
Electron agent/types.ts re-exports from node-runtime.
Added 8 unit tests covering token estimation, tool tracking, usage cloning.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
diguaandCursor committed 2026-05-17 23:32:54 +08:00
1 parent 0ac90030cb
commit 38b599034f
6 files changed
+348 -223

No files matched your search

+9 -189
View File
@@ -1,192 +1,12 @@
/**
* Agent 事件处理器
* 将 AgentCoreEvent 映射为 IPC AgentStreamChunk,
* 并跟踪运行时状态(usage / 工具轮次 / 状态发射 / context token 估算)
* Agent event handler — Electron adapter.
* Re-exports the shared AgentEventHandler from @openchatlab/node-runtime.
*/
import type { PiMessage, AgentCoreEvent } from '@openchatlab/node-runtime'
import type { TokenUsage, AgentRuntimeStatus } from '../../../shared/types'
import type { ToolContext } from '../tools/types'
import type { AgentStreamChunk } from './types'
export interface EventHandlerConfig {
onChunk: (chunk: AgentStreamChunk) => void
context: ToolContext
systemPrompt: string
}
/**
* Agent 运行时状态追踪器
* 可被 Agent 主类作为组合成员持有
*/
export class AgentEventHandler {
readonly toolsUsed: string[] = []
toolRounds: number = 0
private totalUsage: TokenUsage = { promptTokens: 0, completionTokens: 0, totalTokens: 0 }
private lastStatusAt = 0
private readonly onChunk: (chunk: AgentStreamChunk) => void
private readonly context: ToolContext
private readonly systemPrompt: string
constructor(config: EventHandlerConfig) {
this.onChunk = config.onChunk
this.context = config.context
this.systemPrompt = config.systemPrompt
}
/**
* 处理来自 runAgentCore 的事件,映射为 AgentStreamChunk 并发送
* @param event AgentCoreEvent
* @param messages 最近一次 convertToLlm 的 PiMessage 快照(用于 context token 估算)
*/
handleCoreEvent(event: AgentCoreEvent, messages: PiMessage[]): void {
switch (event.type) {
case 'content':
this.onChunk({ type: 'content', content: event.content })
this.emitStatus('responding', messages)
break
case 'thinking_start':
this.emitStatus('thinking', messages, { force: true })
break
case 'thinking_delta':
this.onChunk({ type: 'think', content: event.content, thinkTag: 'thinking' })
this.emitStatus('thinking', messages)
break
case 'thinking_end':
this.onChunk({ type: 'think', content: '', thinkTag: 'thinking', thinkDurationMs: event.durationMs })
this.emitStatus('responding', messages, { force: true })
break
case 'tool_start': {
const params = this.normalizeToolParams(event.toolName, event.toolParams)
this.toolsUsed.push(event.toolName)
this.onChunk({ type: 'tool_start', toolName: event.toolName, toolParams: params })
this.emitStatus('tool_running', messages, { currentTool: event.toolName, force: true })
break
}
case 'tool_end':
this.onChunk({ type: 'tool_result', toolName: event.toolName, toolResult: event.toolResult })
this.emitStatus('thinking', messages, { force: true })
break
case 'turn_end':
this.toolRounds = event.round
this.emitStatus('thinking', messages, { force: true })
break
case 'usage_update':
this.totalUsage = { ...event.usage }
this.emitStatus('responding', messages, { force: true })
break
}
}
cloneUsage(): TokenUsage {
return {
promptTokens: this.totalUsage.promptTokens,
completionTokens: this.totalUsage.completionTokens,
totalTokens: this.totalUsage.totalTokens,
}
}
emitStatus(
phase: AgentRuntimeStatus['phase'],
messages: PiMessage[],
options?: {
pendingUserMessage?: string
currentTool?: string
force?: boolean
}
): void {
const now = Date.now()
if (!options?.force && now - this.lastStatusAt < 240) {
return
}
this.lastStatusAt = now
const contextTokens = this.estimateContextTokens(this.systemPrompt, messages, options?.pendingUserMessage)
const status: AgentRuntimeStatus = {
phase,
round: this.toolRounds,
toolsUsed: this.toolsUsed.length,
currentTool: options?.currentTool,
contextTokens,
totalUsage: this.cloneUsage(),
updatedAt: now,
}
this.onChunk({ type: 'status', status })
}
normalizeToolParams(toolName: string, params: Record<string, unknown>): Record<string, unknown> {
const normalized = { ...params }
const toolsWithLimit = ['search_messages', 'get_recent_messages', 'get_conversation_between']
if (this.context.maxMessagesLimit && toolsWithLimit.includes(toolName)) {
normalized.limit = this.context.maxMessagesLimit
}
if (this.context.timeFilter && (toolName === 'search_messages' || toolName === 'get_recent_messages')) {
normalized._timeFilter = this.context.timeFilter
}
return normalized
}
// ==================== Token 估算(内部方法) ====================
private estimateTokensFromText(text: string): number {
if (!text) return 0
const normalized = text.replace(/\s+/g, ' ').trim()
if (!normalized) return 0
const cjkCount = (normalized.match(/[\u3400-\u9fff\uf900-\ufaff]/g) || []).length
const latinCount = normalized.length - cjkCount
return Math.max(1, Math.ceil(cjkCount * 1.15 + latinCount / 4))
}
private extractMessageText(message: PiMessage): string {
if (message.role === 'user') {
if (typeof message.content === 'string') return message.content
return message.content
.map((item) => {
if (item.type === 'text') return item.text
if (item.type === 'image') return '[image]'
return ''
})
.join('\n')
}
if (message.role === 'assistant') {
return message.content
.map((item) => {
if (item.type === 'text') return item.text
if (item.type === 'thinking') return item.thinking
if (item.type === 'toolCall') return `${item.name} ${JSON.stringify(item.arguments || {})}`
return ''
})
.join('\n')
}
if (message.role === 'toolResult') {
return message.content
.map((item) => {
if (item.type === 'text') return item.text
return '[binary]'
})
.join('\n')
}
return ''
}
private estimateContextTokens(systemPrompt: string, messages: PiMessage[], pendingUserMessage?: string): number {
let tokens = this.estimateTokensFromText(systemPrompt)
for (const message of messages) {
if (message.role === 'toolResult') continue
tokens += this.estimateTokensFromText(this.extractMessageText(message))
}
if (pendingUserMessage) {
tokens += this.estimateTokensFromText(pendingUserMessage)
}
return tokens
}
}
export {
AgentEventHandler,
type EventHandlerConfig,
type TokenUsage,
type AgentRuntimeStatus,
type AgentStreamChunk,
} from '@openchatlab/node-runtime'
+6 -34
View File
@@ -2,9 +2,11 @@
* Agent 模块类型定义
*/
import type { TokenUsage, AgentRuntimeStatus, SerializedErrorInfo } from '../../../shared/types'
import type { TokenUsage } from '@openchatlab/node-runtime'
import type { SerializedErrorInfo } from '../../../shared/types'
export type { TokenUsage, AgentRuntimeStatus, SerializedErrorInfo } from '../../../shared/types'
export type { TokenUsage, AgentRuntimeStatus } from '@openchatlab/node-runtime'
export type { SerializedErrorInfo }
/**
* Agent 配置
@@ -17,39 +19,9 @@ export interface AgentConfig {
}
/**
* Agent 流式响应 chunk
* Agent stream chunk — re-exported from shared, used by Electron IPC layer.
*/
export interface AgentStreamChunk {
/** chunk 类型 */
type: 'content' | 'think' | 'tool_start' | 'tool_result' | 'status' | 'compression_done' | 'done' | 'error'
/** 文本内容(type=content 时) */
content?: string
/** 思考标签名称(type=think 时) */
thinkTag?: string
/** 思考耗时(毫秒,type=think 时可选) */
thinkDurationMs?: number
/** 工具名称(type=tool_start/tool_result 时) */
toolName?: string
/** 工具调用参数(type=tool_start 时) */
toolParams?: Record<string, unknown>
/** 工具执行结果(type=tool_result 时) */
toolResult?: unknown
/** 结构化错误信息(type=error 时) */
error?: SerializedErrorInfo
/** 是否完成 */
isFinished?: boolean
/** Token 使用量(type=done 时返回累计值) */
usage?: TokenUsage
/** 运行状态(type=status 时返回) */
status?: AgentRuntimeStatus
/** 压缩结果(type=compression_done 时) */
compressionResult?: {
summaryContent: string
tokensBefore: number
tokensAfter: number
timestamp: number
}
}
export type { AgentStreamChunk } from '@openchatlab/node-runtime'
/**
* Agent 执行结果
@@ -0,0 +1,100 @@
/**
* Tests for shared AgentEventHandler.
*
* Run: npx tsx --test packages/node-runtime/src/ai/agent/__tests__/event-handler.test.ts
*/
import { describe, it } from 'node:test'
import assert from 'node:assert/strict'
import { AgentEventHandler, estimateTokensFromText } from '../event-handler'
import type { AgentStreamChunk } from '../event-handler'
describe('estimateTokensFromText', () => {
it('returns 0 for empty text', () => {
assert.equal(estimateTokensFromText(''), 0)
})
it('estimates latin text at ~4 chars per token', () => {
const estimate = estimateTokensFromText('hello world, this is a test')
assert.ok(estimate > 0)
assert.ok(estimate < 20)
})
it('estimates CJK text at ~1 char per token', () => {
const estimate = estimateTokensFromText('你好世界这是测试')
assert.ok(estimate >= 8)
})
})
describe('AgentEventHandler', () => {
it('tracks tool usage on tool_start events', () => {
const chunks: AgentStreamChunk[] = []
const handler = new AgentEventHandler({
onChunk: (c) => chunks.push(c),
context: {},
systemPrompt: 'test',
})
handler.handleCoreEvent({ type: 'tool_start', toolName: 'search', toolParams: { q: 'test' } }, [])
assert.deepEqual(handler.toolsUsed, ['search'])
})
it('updates tool rounds on turn_end', () => {
const handler = new AgentEventHandler({
onChunk: () => {},
context: {},
systemPrompt: 'test',
})
handler.handleCoreEvent({ type: 'turn_end', round: 3, hadToolCalls: true }, [])
assert.equal(handler.toolRounds, 3)
})
it('cloneUsage returns independent copy', () => {
const handler = new AgentEventHandler({
onChunk: () => {},
context: {},
systemPrompt: 'test',
})
handler.handleCoreEvent(
{ type: 'usage_update', usage: { promptTokens: 100, completionTokens: 50, totalTokens: 150 } },
[]
)
const usage = handler.cloneUsage()
assert.equal(usage.totalTokens, 150)
handler.handleCoreEvent(
{ type: 'usage_update', usage: { promptTokens: 200, completionTokens: 100, totalTokens: 300 } },
[]
)
assert.equal(usage.totalTokens, 150, 'clone should be independent')
})
it('normalizes tool params with context limits', () => {
const handler = new AgentEventHandler({
onChunk: () => {},
context: { maxMessagesLimit: 10, timeFilter: { startTs: 100, endTs: 200 } },
systemPrompt: 'test',
})
const params = handler.normalizeToolParams('search_messages', { query: 'test' })
assert.equal(params.limit, 10)
assert.deepEqual(params._timeFilter, { startTs: 100, endTs: 200 })
})
it('emits content chunks', () => {
const chunks: AgentStreamChunk[] = []
const handler = new AgentEventHandler({
onChunk: (c) => chunks.push(c),
context: {},
systemPrompt: 'test',
})
handler.handleCoreEvent({ type: 'content', content: 'Hello' }, [])
const contentChunks = chunks.filter((c) => c.type === 'content')
assert.equal(contentChunks.length, 1)
assert.equal(contentChunks[0].content, 'Hello')
})
})
@@ -0,0 +1,219 @@
/**
* Agent event handler — shared implementation.
*
* Maps AgentCoreEvent → stream chunks, tracks usage/tool rounds,
* estimates context tokens. Platform-agnostic via generic chunk type.
*/
import type { Message as PiMessage } from '@mariozechner/pi-ai'
import type { AgentCoreEvent } from './types'
// ==================== Shared types ====================
export interface TokenUsage {
promptTokens: number
completionTokens: number
totalTokens: number
}
export interface AgentRuntimeStatus {
phase: 'compressing' | 'preparing' | 'thinking' | 'tool_running' | 'responding' | 'completed' | 'aborted' | 'error'
round: number
toolsUsed: number
currentTool?: string
contextTokens: number
totalUsage: TokenUsage
updatedAt: number
}
export interface AgentStreamChunk {
type: 'content' | 'think' | 'tool_start' | 'tool_result' | 'status' | 'compression_done' | 'done' | 'error'
content?: string
thinkTag?: string
thinkDurationMs?: number
toolName?: string
toolParams?: Record<string, unknown>
toolResult?: unknown
error?: unknown
isFinished?: boolean
usage?: TokenUsage
status?: AgentRuntimeStatus
compressionResult?: {
summaryContent: string
tokensBefore: number
tokensAfter: number
timestamp: number
}
}
export interface EventHandlerContext {
maxMessagesLimit?: number
timeFilter?: { startTs: number; endTs: number }
}
export interface EventHandlerConfig {
onChunk: (chunk: AgentStreamChunk) => void
context: EventHandlerContext
systemPrompt: string
}
// ==================== Token estimation ====================
function estimateTokensFromText(text: string): number {
if (!text) return 0
const normalized = text.replace(/\s+/g, ' ').trim()
if (!normalized) return 0
const cjkCount = (normalized.match(/[\u3400-\u9fff\uf900-\ufaff]/g) || []).length
const latinCount = normalized.length - cjkCount
return Math.max(1, Math.ceil(cjkCount * 1.15 + latinCount / 4))
}
function extractMessageText(message: PiMessage): string {
if (message.role === 'user') {
if (typeof message.content === 'string') return message.content
return message.content
.map((item) => {
if (item.type === 'text') return item.text
if (item.type === 'image') return '[image]'
return ''
})
.join('\n')
}
if (message.role === 'assistant') {
return message.content
.map((item) => {
if (item.type === 'text') return item.text
if (item.type === 'thinking') return item.thinking
if (item.type === 'toolCall') return `${item.name} ${JSON.stringify(item.arguments || {})}`
return ''
})
.join('\n')
}
if (message.role === 'toolResult') {
return message.content
.map((item) => {
if (item.type === 'text') return item.text
return '[binary]'
})
.join('\n')
}
return ''
}
function estimateContextTokens(systemPrompt: string, messages: PiMessage[], pendingUserMessage?: string): number {
let tokens = estimateTokensFromText(systemPrompt)
for (const message of messages) {
if (message.role === 'toolResult') continue
tokens += estimateTokensFromText(extractMessageText(message))
}
if (pendingUserMessage) {
tokens += estimateTokensFromText(pendingUserMessage)
}
return tokens
}
// Exported for unit testing
export { estimateTokensFromText }
// ==================== Event handler class ====================
export class AgentEventHandler {
readonly toolsUsed: string[] = []
toolRounds: number = 0
private totalUsage: TokenUsage = { promptTokens: 0, completionTokens: 0, totalTokens: 0 }
private lastStatusAt = 0
private readonly onChunk: (chunk: AgentStreamChunk) => void
private readonly context: EventHandlerContext
private readonly systemPrompt: string
constructor(config: EventHandlerConfig) {
this.onChunk = config.onChunk
this.context = config.context
this.systemPrompt = config.systemPrompt
}
handleCoreEvent(event: AgentCoreEvent, messages: PiMessage[]): void {
switch (event.type) {
case 'content':
this.onChunk({ type: 'content', content: event.content })
this.emitStatus('responding', messages)
break
case 'thinking_start':
this.emitStatus('thinking', messages, { force: true })
break
case 'thinking_delta':
this.onChunk({ type: 'think', content: event.content, thinkTag: 'thinking' })
this.emitStatus('thinking', messages)
break
case 'thinking_end':
this.onChunk({ type: 'think', content: '', thinkTag: 'thinking', thinkDurationMs: event.durationMs })
this.emitStatus('responding', messages, { force: true })
break
case 'tool_start': {
const params = this.normalizeToolParams(event.toolName, event.toolParams)
this.toolsUsed.push(event.toolName)
this.onChunk({ type: 'tool_start', toolName: event.toolName, toolParams: params })
this.emitStatus('tool_running', messages, { currentTool: event.toolName, force: true })
break
}
case 'tool_end':
this.onChunk({ type: 'tool_result', toolName: event.toolName, toolResult: event.toolResult })
this.emitStatus('thinking', messages, { force: true })
break
case 'turn_end':
this.toolRounds = event.round
this.emitStatus('thinking', messages, { force: true })
break
case 'usage_update':
this.totalUsage = { ...event.usage }
this.emitStatus('responding', messages, { force: true })
break
}
}
cloneUsage(): TokenUsage {
return {
promptTokens: this.totalUsage.promptTokens,
completionTokens: this.totalUsage.completionTokens,
totalTokens: this.totalUsage.totalTokens,
}
}
emitStatus(
phase: AgentRuntimeStatus['phase'],
messages: PiMessage[],
options?: { pendingUserMessage?: string; currentTool?: string; force?: boolean }
): void {
const now = Date.now()
if (!options?.force && now - this.lastStatusAt < 240) return
this.lastStatusAt = now
const contextTokens = estimateContextTokens(this.systemPrompt, messages, options?.pendingUserMessage)
const status: AgentRuntimeStatus = {
phase,
round: this.toolRounds,
toolsUsed: this.toolsUsed.length,
currentTool: options?.currentTool,
contextTokens,
totalUsage: this.cloneUsage(),
updatedAt: now,
}
this.onChunk({ type: 'status', status })
}
normalizeToolParams(toolName: string, params: Record<string, unknown>): Record<string, unknown> {
const normalized = { ...params }
const toolsWithLimit = ['search_messages', 'get_recent_messages', 'get_conversation_between']
if (this.context.maxMessagesLimit && toolsWithLimit.includes(toolName)) {
normalized.limit = this.context.maxMessagesLimit
}
if (this.context.timeFilter && (toolName === 'search_messages' || toolName === 'get_recent_messages')) {
normalized._timeFilter = this.context.timeFilter
}
return normalized
}
}
+10
View File
@@ -62,6 +62,16 @@ export type { PreprocessingPipelineOptions, PreprocessingPipelineResult } from '
export type { AgentCoreOptions, AgentCoreEvent, AgentCoreResult, AgentTokenUsage, SimpleHistoryMessage } from './agent'
export { runAgentCore } from './agent'
// Agent Event Handler
export { AgentEventHandler, estimateTokensFromText } from './agent/event-handler'
export type {
TokenUsage,
AgentRuntimeStatus,
AgentStreamChunk,
EventHandlerConfig,
EventHandlerContext,
} from './agent/event-handler'
// Agent Prompt Builder
export { buildSystemPrompt } from './agent/prompt-builder'
export type {
+4
View File
@@ -88,6 +88,10 @@ export type {
export type { AgentCoreOptions, AgentCoreEvent, AgentCoreResult, AgentTokenUsage, SimpleHistoryMessage } from './ai'
export { runAgentCore } from './ai'
// Agent Event Handler
export { AgentEventHandler, estimateTokensFromText } from './ai'
export type { TokenUsage, AgentRuntimeStatus, AgentStreamChunk, EventHandlerConfig, EventHandlerContext } from './ai'
// Agent Prompt Builder
export { buildSystemPrompt } from './ai'
export type { BuildSystemPromptOptions, OwnerInfo, MentionedMember, SkillContext, TranslateFn } from './ai'