| 1 | /** |
| 2 | * TaskInstance - SSE 消息处理模块 |
| 3 | * 处理来自服务端的 SSE 事件 |
| 4 | */ |
| 5 | |
| 6 | import type { |
| 7 | IActionCard, |
| 8 | IDisplayMessage, |
| 9 | IMediaItem, |
| 10 | IPublishFlowData, |
| 11 | ISSEMessage, |
| 12 | IWorkflowStep, |
| 13 | } from '../agent.types' |
| 14 | import type { ISSECallbacks, ISSEHandlerContext } from './task-instance.types' |
| 15 | import { |
| 16 | addMarkdownMessage, |
| 17 | updateMessageContent, |
| 18 | updateMessageContentWithMedias, |
| 19 | updateMessageWithActionsAndPublishFlows, |
| 20 | } from './message.handler' |
| 21 | import { |
| 22 | addWorkflowStep, |
| 23 | handleTextDelta, |
| 24 | handleToolCallComplete, |
| 25 | handleToolResult, |
| 26 | saveCurrentStepToMessage, |
| 27 | startNewStep, |
| 28 | updateLastWorkflowStep, |
| 29 | } from './workflow.handler' |
| 30 | |
| 31 | // ============ SSE 消息处理主入口 ============ |
| 32 | |
| 33 | /** |
| 34 | * 处理 SSE 消息 |
| 35 | * 所有消息都会写入到当前任务数据中 |
| 36 | */ |
| 37 | export function handleSSEMessage( |
| 38 | ctx: ISSEHandlerContext, |
| 39 | msg: ISSEMessage, |
| 40 | callbacks?: ISSECallbacks, |
| 41 | ): void { |
| 42 | // 处理 init 消息 - 迁移到真实 taskId |
| 43 | if (msg.type === 'init' && msg.taskId) { |
| 44 | ctx.migrateToRealTaskId(msg.taskId) |
| 45 | ctx.setStreamingText('') |
| 46 | callbacks?.onTaskIdReady?.(msg.taskId) |
| 47 | return |
| 48 | } |
| 49 | |
| 50 | // 心跳消息，忽略 |
| 51 | if (msg.type === 'keep_alive') { |
| 52 | return |
| 53 | } |
| 54 | |
| 55 | // 处理 stream_event 消息 |
| 56 | if (msg.type === 'stream_event') { |
| 57 | handleStreamEvent(ctx, msg) |
| 58 | return |
| 59 | } |
| 60 | |
| 61 | // 处理 assistant 消息（工具调用完成） |
| 62 | if (msg.type === 'assistant' && msg.message) { |
| 63 | handleAssistantMessage(ctx, msg) |
| 64 | return |
| 65 | } |
| 66 | |
| 67 | // 处理 user 消息（工具结果） |
| 68 | if (msg.type === 'user' && msg.message) { |
| 69 | handleUserMessage(ctx, msg) |
| 70 | return |
| 71 | } |
| 72 | |
| 73 | // 处理 text 消息 |
| 74 | if (msg.type === 'text' && msg.message) { |
| 75 | addMarkdownMessage(ctx, msg.message as string) |
| 76 | return |
| 77 | } |
| 78 | |
| 79 | // 处理 result 消息 |
| 80 | if (msg.type === 'result') { |
| 81 | handleResultMessage(ctx, msg) |
| 82 | return |
| 83 | } |
| 84 | |
| 85 | // 处理 error 消息 |
| 86 | if (msg.type === 'error') { |
| 87 | handleErrorMessage(ctx, msg, callbacks) |
| 88 | return |
| 89 | } |
| 90 | |
| 91 | // 处理 done 消息 |
| 92 | if (msg.type === 'done') { |
| 93 | // 先保存当前步骤（如果有内容），防止消息丢失 |
| 94 | const streamingText = ctx.getStreamingText() |
| 95 | const currentStepWorkflow = ctx.getCurrentStepWorkflow() |
| 96 | if (streamingText.trim() || currentStepWorkflow.length > 0) { |
| 97 | saveCurrentStepToMessage(ctx) |
| 98 | } |
| 99 | |
| 100 | ctx.markMessageDone() |
| 101 | callbacks?.onComplete?.() |
| 102 | } |
| 103 | } |
| 104 | |
| 105 | // ============ 内部处理函数 ============ |
| 106 | |
| 107 | /** |
| 108 | * 处理 stream_event 消息 |
| 109 | */ |
| 110 | function handleStreamEvent(ctx: ISSEHandlerContext, msg: ISSEMessage): void { |
| 111 | const event = extractEvent(msg) |
| 112 | if (!event) |
| 113 | return |
| 114 | |
| 115 | // message_start - 开始新步骤 |
| 116 | // 每个新的"消息"（包含文本或工具调用）都会有一个 message_start 事件 |
| 117 | // 只有当有文字内容时才开始新步骤，连续的MCP调用（没有文字消息间隔）应该属于同一个步骤 |
| 118 | if (event.type === 'message_start') { |
| 119 | // 只有当有文字内容时才开始新步骤 |
| 120 | // 连续的MCP调用（没有文字消息间隔）应该属于同一个步骤 |
| 121 | const streamingText = ctx.getStreamingText() |
| 122 | |
| 123 | if (streamingText.trim()) { |
| 124 | startNewStep(ctx) |
| 125 | } |
| 126 | else if (ctx.getCurrentStepIndex() < 0) { |
| 127 | // 第一次 message_start，初始化 stepIndex 为 0 |
| 128 | ctx.incrementCurrentStepIndex() |
| 129 | } |
| 130 | |
| 131 | return |
| 132 | } |
| 133 | |
| 134 | // content_block_start (tool_use) |
| 135 | if (event.type === 'content_block_start' && event.content_block?.type === 'tool_use') { |
| 136 | const toolName = event.content_block.name || 'Unknown Tool' |
| 137 | const toolId = event.content_block.id || `tool-${Date.now()}` |
| 138 | |
| 139 | const newStep: IWorkflowStep = { |
| 140 | id: toolId, |
| 141 | type: 'tool_call', |
| 142 | toolName, |
| 143 | content: '', |
| 144 | isActive: true, |
| 145 | timestamp: Date.now(), |
| 146 | } |
| 147 | addWorkflowStep(ctx, newStep) |
| 148 | return |
| 149 | } |
| 150 | |
| 151 | // text_delta - 追加文本，并检查是否需要分割步骤 |
| 152 | if (event.type === 'content_block_delta' && event.delta?.type === 'text_delta') { |
| 153 | const text = event.delta.text |
| 154 | if (!text) |
| 155 | return |
| 156 | |
| 157 | // 如果当前步骤已经有工作流但没有文字，开始新步骤 |
| 158 | // 这样开头的 MCP 调用会在一个步骤，文字开始后是新步骤 |
| 159 | const currentStepWorkflow = ctx.getCurrentStepWorkflow() |
| 160 | const streamingText = ctx.getStreamingText() |
| 161 | if (currentStepWorkflow.length > 0 && !streamingText.trim()) { |
| 162 | startNewStep(ctx) |
| 163 | } |
| 164 | |
| 165 | ctx.appendStreamingText(text) |
| 166 | handleTextDelta(ctx) |
| 167 | return |
| 168 | } |
| 169 | |
| 170 | // input_json_delta |
| 171 | if (event.type === 'content_block_delta' && event.delta?.type === 'input_json_delta') { |
| 172 | const partialJson = event.delta.partial_json |
| 173 | if (partialJson) { |
| 174 | updateLastWorkflowStep(ctx, step => ({ |
| 175 | ...step, |
| 176 | content: (step.content || '') + partialJson, |
| 177 | })) |
| 178 | } |
| 179 | } |
| 180 | } |
| 181 | |
| 182 | /** |
| 183 | * 处理 assistant 消息 |
| 184 | */ |
| 185 | function handleAssistantMessage(ctx: ISSEHandlerContext, msg: ISSEMessage): void { |
| 186 | const assistantMsg = msg.message as any |
| 187 | if (assistantMsg?.message?.content && Array.isArray(assistantMsg.message.content)) { |
| 188 | assistantMsg.message.content.forEach((item: any) => { |
| 189 | if (item.type === 'tool_use') { |
| 190 | const toolName = item.name || 'Unknown Tool' |
| 191 | const toolInput = item.input ? JSON.stringify(item.input, null, 2) : '' |
| 192 | handleToolCallComplete(ctx, toolName, toolInput) |
| 193 | } |
| 194 | }) |
| 195 | } |
| 196 | } |
| 197 | |
| 198 | /** |
| 199 | * 处理 user 消息（工具结果） |
| 200 | */ |
| 201 | function handleUserMessage(ctx: ISSEHandlerContext, msg: ISSEMessage): void { |
| 202 | const userMsg = msg.message as any |
| 203 | const contentArray = userMsg?.content || userMsg?.message?.content |
| 204 | if (contentArray && Array.isArray(contentArray)) { |
| 205 | contentArray.forEach((item: any) => { |
| 206 | if (item.type === 'tool_result') { |
| 207 | let resultText = '' |
| 208 | if (Array.isArray(item.content)) { |
| 209 | item.content.forEach((rc: any) => { |
| 210 | if (rc.type === 'text') { |
| 211 | resultText = rc.text || '' |
| 212 | } |
| 213 | }) |
| 214 | } |
| 215 | else if (typeof item.content === 'string') { |
| 216 | resultText = item.content |
| 217 | } |
| 218 | if (resultText) { |
| 219 | handleToolResult(ctx, resultText) |
| 220 | } |
| 221 | } |
| 222 | }) |
| 223 | } |
| 224 | } |
| 225 | |
| 226 | /** |
| 227 | * 处理 result 消息 |
| 228 | * 支持 result 为数组的情况（多平台发布） |
| 229 | */ |
| 230 | function handleResultMessage(ctx: ISSEHandlerContext, msg: ISSEMessage): void { |
| 231 | const messageData = msg.data || msg.message |
| 232 | if (!messageData) |
| 233 | return |
| 234 | |
| 235 | // 获取 result 数组：优先从 messageData.result 获取，否则将 messageData 视为单个结果 |
| 236 | const resultArray: any[] = Array.isArray(messageData.result) ? messageData.result : [messageData] |
| 237 | |
| 238 | const actions: IActionCard[] = [] |
| 239 | const publishFlows: IPublishFlowData[] = [] |
| 240 | |
| 241 | // 优先使用流式传输累积的完整内容 |
| 242 | // 如果 streamingText 为空（被 startNewStep 清空），content 保持为空 |
| 243 | // 更新函数会保留消息原有的 content，避免被 description 覆盖截断 |
| 244 | const streamingText = ctx.getStreamingText() |
| 245 | const content = streamingText.trim() |
| 246 | |
| 247 | // 遍历 result 数组处理每个结果 |
| 248 | for (const data of resultArray) { |
| 249 | // 如果有 flowId，创建发布流程数据 |
| 250 | if (data.flowId) { |
| 251 | publishFlows.push({ |
| 252 | flowId: data.flowId, |
| 253 | platform: data.platform, |
| 254 | initialData: { |
| 255 | title: data.title, |
| 256 | description: data.description, |
| 257 | medias: data.medias, |
| 258 | }, |
| 259 | }) |
| 260 | } |
| 261 | |
| 262 | // 如果有 action，创建 action 卡片 |
| 263 | if (data.action) { |
| 264 | actions.push({ |
| 265 | type: data.action, |
| 266 | platform: data.platform, |
| 267 | accountId: data.accountId, |
| 268 | title: data.title, |
| 269 | description: data.description, |
| 270 | medias: data.medias as IMediaItem[], |
| 271 | tags: data.tags || data.topics, |
| 272 | flowId: data.flowId, |
| 273 | _isRealtime: true, |
| 274 | }) |
| 275 | } |
| 276 | } |
| 277 | |
| 278 | // 更新消息 |
| 279 | if (publishFlows.length > 0 || actions.length > 0) { |
| 280 | updateMessageWithActionsAndPublishFlows(ctx, content, actions, publishFlows) |
| 281 | } |
| 282 | else { |
| 283 | // 检查第一个结果是否有媒体 |
| 284 | const firstData = resultArray[0] |
| 285 | if (firstData?.medias && firstData.medias.length > 0) { |
| 286 | updateMessageContentWithMedias(ctx, content, firstData.medias) |
| 287 | } |
| 288 | else if (content) { |
| 289 | updateMessageContent(ctx, content) |
| 290 | } |
| 291 | } |
| 292 | } |
| 293 | |
| 294 | /** |
| 295 | * 处理 error 消息 |
| 296 | */ |
| 297 | async function handleErrorMessage( |
| 298 | ctx: ISSEHandlerContext, |
| 299 | msg: ISSEMessage, |
| 300 | callbacks?: ISSECallbacks, |
| 301 | ): Promise<void> { |
| 302 | const errorCode = (msg as any).code |
| 303 | const errorMessage |
| 304 | = typeof msg.message === 'string' ? msg.message : (msg.message as any)?.message || 'Unknown error' |
| 305 | |
| 306 | if (errorCode === 12001) { |
| 307 | // 先保存当前步骤内容，防止正在流式传输的文字消息丢失 |
| 308 | const streamingText = ctx.getStreamingText() |
| 309 | const currentStepWorkflow = ctx.getCurrentStepWorkflow() |
| 310 | if (streamingText.trim() || currentStepWorkflow.length > 0) { |
| 311 | saveCurrentStepToMessage(ctx) |
| 312 | } |
| 313 | |
| 314 | // 积分不足：在聊天中显示卡片，不跳转 |
| 315 | const insufficientMsg: IDisplayMessage = { |
| 316 | id: `assistant-insufficient-${Date.now()}`, |
| 317 | role: 'assistant', |
| 318 | content: '', |
| 319 | status: 'done', |
| 320 | createdAt: Date.now(), |
| 321 | actions: [ |
| 322 | { |
| 323 | type: 'insufficientCredits', |
| 324 | }, |
| 325 | ], |
| 326 | } |
| 327 | ctx.addMessage(insufficientMsg) |
| 328 | } |
| 329 | else { |
| 330 | // 其他错误：创建错误消息 |
| 331 | if (errorMessage) { |
| 332 | const errorMsg: IDisplayMessage = { |
| 333 | id: `assistant-error-${Date.now()}`, |
| 334 | role: 'assistant', |
| 335 | content: '', |
| 336 | status: 'done', |
| 337 | createdAt: Date.now(), |
| 338 | actions: [ |
| 339 | { |
| 340 | type: 'errorOnly', |
| 341 | title: '生成失败', |
| 342 | description: errorMessage, |
| 343 | }, |
| 344 | ], |
| 345 | } |
| 346 | ctx.addMessage(errorMsg) |
| 347 | } |
| 348 | } |
| 349 | |
| 350 | // 更新任务状态 |
| 351 | setTimeout(() => { |
| 352 | ctx.setIsGenerating(false) |
| 353 | }, 100) |
| 354 | ctx.setProgress(0) |
| 355 | |
| 356 | callbacks?.onError?.(new Error(errorMessage)) |
| 357 | } |
| 358 | |
| 359 | // ============ 工具函数 ============ |
| 360 | |
| 361 | /** |
| 362 | * 从 SSE 消息中提取 event 对象 |
| 363 | */ |
| 364 | function extractEvent(msg: ISSEMessage): any { |
| 365 | if ((msg as any).event) { |
| 366 | return (msg as any).event |
| 367 | } |
| 368 | if (msg.message && typeof msg.message === 'object') { |
| 369 | return (msg.message as any).event |
| 370 | } |
| 371 | return null |
| 372 | } |
| 373 |