feat: add MaiSaka real-time chat flow monitoring component and WebSocket event handling
- Implemented the MaiSakaMonitor component for real-time monitoring of chat flow using WebSocket. - Created a custom hook `useMaisakaMonitor` to manage WebSocket subscriptions and event states. - Developed a backend module for broadcasting various monitoring events through WebSocket. - Added serialization functions for messages and tool calls to optimize data transmission. - Included event emission functions for session start, message ingestion, cycle start, timing gate results, planner requests/responses, tool executions, and replier requests/responses.
This commit is contained in:
DrSmoothl
@@ -30,7 +30,7 @@ export async function getApiBaseUrl(): Promise<string> {
|
||||
/**
|
||||
* Get WebSocket base URL
|
||||
* - Electron: Convert HTTP/HTTPS URL to WS/WSS
|
||||
* - Browser DEV: ws://127.0.0.1:8001 (hardcoded, same as log-websocket.ts)
|
||||
* - Browser DEV: Use same-origin WS URL and let Vite proxy forward requests
|
||||
* - Browser PROD: Construct WS URL from window.location
|
||||
*/
|
||||
export async function getWsBaseUrl(): Promise<string> {
|
||||
@@ -47,9 +47,10 @@ export async function getWsBaseUrl(): Promise<string> {
|
||||
})
|
||||
}
|
||||
|
||||
// Browser DEV: Use hardcoded WebSocket server
|
||||
// Browser DEV: Use same-origin URL so Vite proxy can forward WebSocket requests
|
||||
if (import.meta.env.DEV) {
|
||||
return 'ws://127.0.0.1:8001'
|
||||
const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:'
|
||||
return `${protocol}//${window.location.host}`
|
||||
}
|
||||
|
||||
// Browser PROD: Construct WS URL from current location
|
||||
|
||||
223
dashboard/src/lib/maisaka-monitor-client.ts
Normal file
223
dashboard/src/lib/maisaka-monitor-client.ts
Normal file
@@ -0,0 +1,223 @@
|
||||
/**
|
||||
* MaiSaka 实时监控 WebSocket 客户端
|
||||
*
|
||||
* 订阅 maisaka_monitor 主题，接收推理引擎各阶段的实时事件。
|
||||
*/
|
||||
import type { WsEventEnvelope } from './unified-ws'
|
||||
|
||||
import { unifiedWsClient } from './unified-ws'
|
||||
|
||||
// ─── 事件数据类型 ───────────────────────────────────────────────
|
||||
|
||||
export interface MaisakaMessage {
|
||||
role: string
|
||||
content: string | null
|
||||
tool_call_id?: string
|
||||
tool_calls?: MaisakaToolCall[]
|
||||
}
|
||||
|
||||
export interface MaisakaToolCall {
|
||||
id: string
|
||||
name: string
|
||||
arguments?: Record<string, unknown>
|
||||
arguments_raw?: string
|
||||
}
|
||||
|
||||
export interface SessionStartEvent {
|
||||
session_id: string
|
||||
session_name: string
|
||||
timestamp: number
|
||||
}
|
||||
|
||||
export interface MessageIngestedEvent {
|
||||
session_id: string
|
||||
speaker_name: string
|
||||
content: string
|
||||
message_id: string
|
||||
timestamp: number
|
||||
}
|
||||
|
||||
export interface CycleStartEvent {
|
||||
session_id: string
|
||||
cycle_id: number
|
||||
round_index: number
|
||||
max_rounds: number
|
||||
history_count: number
|
||||
timestamp: number
|
||||
}
|
||||
|
||||
export interface TimingGateResultEvent {
|
||||
session_id: string
|
||||
cycle_id: number
|
||||
action: 'continue' | 'wait' | 'no_reply'
|
||||
content: string | null
|
||||
tool_calls: MaisakaToolCall[]
|
||||
messages: MaisakaMessage[]
|
||||
prompt_tokens: number
|
||||
selected_history_count: number
|
||||
duration_ms: number
|
||||
timestamp: number
|
||||
}
|
||||
|
||||
export interface PlannerRequestEvent {
|
||||
session_id: string
|
||||
cycle_id: number
|
||||
messages: MaisakaMessage[]
|
||||
tool_count: number
|
||||
selected_history_count: number
|
||||
timestamp: number
|
||||
}
|
||||
|
||||
export interface PlannerResponseEvent {
|
||||
session_id: string
|
||||
cycle_id: number
|
||||
content: string | null
|
||||
tool_calls: MaisakaToolCall[]
|
||||
prompt_tokens: number
|
||||
completion_tokens: number
|
||||
total_tokens: number
|
||||
duration_ms: number
|
||||
timestamp: number
|
||||
}
|
||||
|
||||
export interface ToolExecutionEvent {
|
||||
session_id: string
|
||||
cycle_id: number
|
||||
tool_name: string
|
||||
tool_args: Record<string, unknown>
|
||||
result_summary: string
|
||||
success: boolean
|
||||
duration_ms: number
|
||||
timestamp: number
|
||||
}
|
||||
|
||||
export interface CycleEndEvent {
|
||||
session_id: string
|
||||
cycle_id: number
|
||||
time_records: Record<string, number>
|
||||
agent_state: string
|
||||
timestamp: number
|
||||
}
|
||||
|
||||
export interface ReplierRequestEvent {
|
||||
session_id: string
|
||||
messages: MaisakaMessage[]
|
||||
model_name: string
|
||||
timestamp: number
|
||||
}
|
||||
|
||||
export interface ReplierResponseEvent {
|
||||
session_id: string
|
||||
content: string | null
|
||||
reasoning: string
|
||||
model_name: string
|
||||
prompt_tokens: number
|
||||
completion_tokens: number
|
||||
total_tokens: number
|
||||
duration_ms: number
|
||||
success: boolean
|
||||
timestamp: number
|
||||
}
|
||||
|
||||
// ─── 统一事件联合类型 ─────────────────────────────────────────
|
||||
|
||||
export type MaisakaMonitorEvent =
|
||||
| { type: 'session.start'; data: SessionStartEvent }
|
||||
| { type: 'message.ingested'; data: MessageIngestedEvent }
|
||||
| { type: 'cycle.start'; data: CycleStartEvent }
|
||||
| { type: 'timing_gate.result'; data: TimingGateResultEvent }
|
||||
| { type: 'planner.request'; data: PlannerRequestEvent }
|
||||
| { type: 'planner.response'; data: PlannerResponseEvent }
|
||||
| { type: 'tool.execution'; data: ToolExecutionEvent }
|
||||
| { type: 'cycle.end'; data: CycleEndEvent }
|
||||
| { type: 'replier.request'; data: ReplierRequestEvent }
|
||||
| { type: 'replier.response'; data: ReplierResponseEvent }
|
||||
|
||||
export type MaisakaEventListener = (event: MaisakaMonitorEvent) => void
|
||||
|
||||
// ─── 客户端 ───────────────────────────────────────────────────
|
||||
|
||||
class MaisakaMonitorClient {
|
||||
private initialized = false
|
||||
private listenerIdCounter = 0
|
||||
private listeners: Map<number, MaisakaEventListener> = new Map()
|
||||
private subscriptionActive = false
|
||||
private subscriptionPromise: Promise<void> | null = null
|
||||
private deferredUnsubTimer: ReturnType<typeof setTimeout> | null = null
|
||||
|
||||
private initialize(): void {
|
||||
if (this.initialized) {
|
||||
return
|
||||
}
|
||||
|
||||
unifiedWsClient.addEventListener((message: WsEventEnvelope) => {
|
||||
if (message.domain !== 'maisaka_monitor') {
|
||||
return
|
||||
}
|
||||
|
||||
const event: MaisakaMonitorEvent = {
|
||||
type: message.event as MaisakaMonitorEvent['type'],
|
||||
data: message.data as never,
|
||||
}
|
||||
|
||||
this.listeners.forEach((listener) => {
|
||||
try {
|
||||
listener(event)
|
||||
} catch (error) {
|
||||
console.error('MaiSaka 监控事件监听器执行失败:', error)
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
this.initialized = true
|
||||
}
|
||||
|
||||
private async ensureSubscribed(): Promise<void> {
|
||||
if (this.subscriptionActive) {
|
||||
return
|
||||
}
|
||||
|
||||
if (this.subscriptionPromise === null) {
|
||||
this.subscriptionPromise = unifiedWsClient
|
||||
.subscribe('maisaka_monitor', 'main')
|
||||
.then(() => {
|
||||
this.subscriptionActive = true
|
||||
})
|
||||
.finally(() => {
|
||||
this.subscriptionPromise = null
|
||||
})
|
||||
}
|
||||
|
||||
await this.subscriptionPromise
|
||||
}
|
||||
|
||||
async subscribe(listener: MaisakaEventListener): Promise<() => Promise<void>> {
|
||||
this.initialize()
|
||||
const listenerId = ++this.listenerIdCounter
|
||||
this.listeners.set(listenerId, listener)
|
||||
|
||||
// 如果有待执行的延迟退订，取消它（React StrictMode 快速卸载/重新挂载）
|
||||
if (this.deferredUnsubTimer !== null) {
|
||||
clearTimeout(this.deferredUnsubTimer)
|
||||
this.deferredUnsubTimer = null
|
||||
}
|
||||
|
||||
await this.ensureSubscribed()
|
||||
|
||||
return async () => {
|
||||
this.listeners.delete(listenerId)
|
||||
if (this.listeners.size === 0 && this.subscriptionActive) {
|
||||
// 延迟退订：等待短暂时间再真正退订，避免 StrictMode 导致的竞态
|
||||
this.deferredUnsubTimer = setTimeout(() => {
|
||||
this.deferredUnsubTimer = null
|
||||
if (this.listeners.size === 0 && this.subscriptionActive) {
|
||||
this.subscriptionActive = false
|
||||
void unifiedWsClient.unsubscribe('maisaka_monitor', 'main')
|
||||
}
|
||||
}, 200)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export const maisakaMonitorClient = new MaisakaMonitorClient()
|
||||
@@ -83,11 +83,14 @@ async function getWsToken(): Promise<string | null> {
|
||||
}
|
||||
|
||||
class UnifiedWebSocketClient {
|
||||
private readonly heartbeatIntervalMs = 30000
|
||||
private readonly heartbeatTimeoutMs = 90000
|
||||
private connectPromise: Promise<void> | null = null
|
||||
private connectionListeners: Set<ConnectionListener> = new Set()
|
||||
private eventListeners: Set<EventListener> = new Set()
|
||||
private hasConnectedOnce = false
|
||||
private heartbeatIntervalId: number | null = null
|
||||
private lastPongAt = 0
|
||||
private manualDisconnect = false
|
||||
private pendingRequests: Map<string, PendingRequest> = new Map()
|
||||
private reconnectAttempts = 0
|
||||
@@ -151,10 +154,21 @@ class UnifiedWebSocketClient {
|
||||
private startHeartbeat(): void {
|
||||
this.stopHeartbeat()
|
||||
this.heartbeatIntervalId = window.setInterval(() => {
|
||||
if (this.ws?.readyState === WebSocket.OPEN) {
|
||||
this.ws.send(JSON.stringify({ op: 'ping' }))
|
||||
if (this.ws?.readyState !== WebSocket.OPEN) {
|
||||
return
|
||||
}
|
||||
}, 30000)
|
||||
|
||||
const now = Date.now()
|
||||
if (this.lastPongAt > 0 && now - this.lastPongAt > this.heartbeatTimeoutMs) {
|
||||
console.warn('统一 WebSocket 心跳超时，准备重连')
|
||||
void this.restart().catch((error) => {
|
||||
console.error('统一 WebSocket 心跳重连失败:', error)
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
this.ws.send(JSON.stringify({ op: 'ping' }))
|
||||
}, this.heartbeatIntervalMs)
|
||||
}
|
||||
|
||||
private clearReconnectTimer(): void {
|
||||
@@ -252,7 +266,11 @@ class UnifiedWebSocketClient {
|
||||
}
|
||||
}
|
||||
|
||||
private handleServerMessage(rawData: string): void {
|
||||
private handleServerMessage(socket: WebSocket, rawData: string): void {
|
||||
if (this.ws !== socket) {
|
||||
return
|
||||
}
|
||||
|
||||
let message: WsServerEnvelope
|
||||
try {
|
||||
message = JSON.parse(rawData) as WsServerEnvelope
|
||||
@@ -262,6 +280,7 @@ class UnifiedWebSocketClient {
|
||||
}
|
||||
|
||||
if (message.op === 'pong') {
|
||||
this.lastPongAt = Date.now()
|
||||
return
|
||||
}
|
||||
|
||||
@@ -297,8 +316,13 @@ class UnifiedWebSocketClient {
|
||||
}
|
||||
}
|
||||
|
||||
private handleClose(event: CloseEvent): void {
|
||||
private handleClose(socket: WebSocket, event: CloseEvent): void {
|
||||
if (this.ws !== socket) {
|
||||
return
|
||||
}
|
||||
|
||||
this.stopHeartbeat()
|
||||
this.lastPongAt = 0
|
||||
this.ws = null
|
||||
this.connectPromise = null
|
||||
this.setStatus('idle')
|
||||
@@ -340,10 +364,16 @@ class UnifiedWebSocketClient {
|
||||
this.ws = socket
|
||||
|
||||
socket.onopen = () => {
|
||||
if (this.ws !== socket) {
|
||||
socket.close()
|
||||
return
|
||||
}
|
||||
|
||||
settled = true
|
||||
const shouldNotifyReconnect = this.hasConnectedOnce
|
||||
this.hasConnectedOnce = true
|
||||
this.reconnectAttempts = 0
|
||||
this.lastPongAt = Date.now()
|
||||
this.startHeartbeat()
|
||||
this.setStatus('connected')
|
||||
resolve()
|
||||
@@ -351,10 +381,14 @@ class UnifiedWebSocketClient {
|
||||
}
|
||||
|
||||
socket.onmessage = (event) => {
|
||||
this.handleServerMessage(event.data)
|
||||
this.handleServerMessage(socket, event.data)
|
||||
}
|
||||
|
||||
socket.onerror = () => {
|
||||
if (this.ws !== socket) {
|
||||
return
|
||||
}
|
||||
|
||||
if (!settled) {
|
||||
settled = true
|
||||
reject(new Error('统一 WebSocket 连接失败'))
|
||||
@@ -366,7 +400,7 @@ class UnifiedWebSocketClient {
|
||||
settled = true
|
||||
reject(new Error(`统一 WebSocket 已关闭 (${event.code})`))
|
||||
}
|
||||
this.handleClose(event)
|
||||
this.handleClose(socket, event)
|
||||
}
|
||||
})
|
||||
})()
|
||||
@@ -384,6 +418,7 @@ class UnifiedWebSocketClient {
|
||||
this.manualDisconnect = true
|
||||
this.clearReconnectTimer()
|
||||
this.stopHeartbeat()
|
||||
this.lastPongAt = 0
|
||||
this.rejectPendingRequests(new Error('统一 WebSocket 已手动断开'))
|
||||
this.connectPromise = null
|
||||
if (this.ws) {
|
||||
|
||||
Reference in New Issue
Block a user