From 6da7b600ff7463854d9fa6b24841696f4f46bdd0 Mon Sep 17 00:00:00 2001 From: Fu Gui Date: Tue, 25 Nov 2025 10:35:01 +0800 Subject: [PATCH 1/6] refactor: modularize server hooks and improve code organization MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Extract server hook logic from src/index.ts to dedicated utils/hooks.ts module - Separate concerns with individual hook setup functions for better maintainability - Add comprehensive JSDoc documentation for all functions - Improve error handling and stream processing organization - Enhance code readability with modular structure 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- src/index.ts | 432 ++++++++++------------------------- src/utils/hooks.ts | 552 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 675 insertions(+), 309 deletions(-) create mode 100644 src/utils/hooks.ts diff --git a/src/index.ts b/src/index.ts index efb38f4e..2fd6d15a 100644 --- a/src/index.ts +++ b/src/index.ts @@ -4,39 +4,43 @@ import { homedir } from "os"; import path, { join } from "path"; import { initConfig, initDir, cleanupLogFiles } from "./utils"; import { createServer } from "./server"; -import { router } from "./utils/router"; -import { apiKeyAuth } from "./middleware/auth"; import { cleanupPidFile, isServiceRunning, savePid, } from "./utils/processCheck"; -import { CONFIG_FILE } from "./constants"; +import { CONFIG_FILE, HOME_DIR } from "./constants"; import { createStream } from 'rotating-file-stream'; -import { HOME_DIR } from "./constants"; -import { sessionUsageCache } from "./utils/cache"; -import {SSEParserTransform} from "./utils/SSEParser.transform"; -import {SSESerializerTransform} from "./utils/SSESerializer.transform"; -import {rewriteStream} from "./utils/rewriteStream"; -import JSON5 from "json5"; -import { IAgent } from "./agents/type"; -import agentsManager from "./agents"; +import { + setupRequestLoggingHook, + setupResponseLoggingHook, + setupAuthHook, + setupAgentAndRoutingHook, + setupErrorEventHook, + setupSendEventHook, + setupAgentProcessingHook, + setupSessionUsageHook, + setupErrorPayloadHook +} from "./utils/hooks"; import { EventEmitter } from "node:events"; const event = new EventEmitter() -async function initializeClaudeConfig() { +/** + * Initialize Claude configuration file + */ +async function initializeClaudeConfig(): Promise { const homeDir = homedir(); const configPath = join(homeDir, ".claude.json"); if (!existsSync(configPath)) { - const userID = Array.from( + const userId = Array.from( { length: 64 }, () => Math.random().toString(16)[2] ).join(""); const configContent = { numStartups: 184, autoUpdaterStatus: "enabled", - userID, + userID: userId, hasCompletedOnboarding: true, lastOnboardingVersion: "1.0.17", projects: {}, @@ -45,89 +49,127 @@ async function initializeClaudeConfig() { } } +/** + * Set up process signal handlers for graceful shutdown + */ +function setupSignalHandlers(): void { + const handleShutdown = (signal: string) => { + console.log(`Received ${signal}, cleaning up...`); + cleanupPidFile(); + process.exit(0); + }; + + process.on("SIGINT", () => handleShutdown("SIGINT")); + process.on("SIGTERM", () => handleShutdown("SIGTERM")); +} + +/** + * Configure logger based on config settings + */ +function configureLogger(config: any, homeDir: string) { + return config.LOG !== false + ? { + level: config.LOG_LEVEL || "debug", + stream: createStream((time, index) => { + if (!time) { + time = new Date(); + } + + const pad = (num: number) => (num > 9 ? "" : "0") + num; + const yearAndMonth = time.getFullYear() + "" + pad(time.getMonth() + 1); + const day = pad(time.getDate()); + const hour = pad(time.getHours()); + const minute = pad(time.getMinutes()); + const second = pad(time.getSeconds()); + + return `./logs/ccr-${yearAndMonth}${day}${hour}${minute}${second}${index ? `_${index}` : ''}.log`; + }, { + path: homeDir, + maxFiles: 3, + interval: "1d", + compress: false, + maxSize: "50M" + }), + } + : false; +} + interface RunOptions { port?: number; } -async function run(options: RunOptions = {}) { - // Check if service is already running - const isRunning = await isServiceRunning() - if (isRunning) { - console.log("✅ Service is already running in the background."); - return; - } - +/** + * Initialize application configuration and directories + */ +async function initializeApp(): Promise { await initializeClaudeConfig(); await initDir(); - // Clean up old log files, keeping only the 10 most recent ones await cleanupLogFiles(); - const config = await initConfig(); - + return await initConfig(); +} - let HOST = config.HOST || "127.0.0.1"; +/** + * Resolve host configuration + */ +function resolveHostConfig(config: any): string { + let host = config.HOST || "127.0.0.1"; if (config.HOST && !config.APIKEY) { - HOST = "127.0.0.1"; + host = "127.0.0.1"; console.warn("⚠️ API key is not set. HOST is forced to 127.0.0.1."); } - const port = config.PORT || 3456; + return host; +} - // Save the PID of the background process - savePid(process.pid); +/** + * Resolve service port configuration + */ +function resolveServicePort(config: any): number { + const defaultPort = config.PORT || 3456; + return process.env.SERVICE_PORT + ? parseInt(process.env.SERVICE_PORT) + : defaultPort; +} - // Handle SIGINT (Ctrl+C) to clean up PID file - process.on("SIGINT", () => { - console.log("Received SIGINT, cleaning up..."); - cleanupPidFile(); - process.exit(0); +/** + * Setup global error handlers for the process + */ +function setupGlobalErrorHandlers(server: any): void { + process.on("uncaughtException", (err) => { + server.logger.error("Uncaught exception:", err); }); - // Handle SIGTERM to clean up PID file - process.on("SIGTERM", () => { - cleanupPidFile(); - process.exit(0); + process.on("unhandledRejection", (reason, promise) => { + server.logger.error("Unhandled rejection at:", promise, "reason:", reason); }); +} - // Use port from environment variable if set (for background process) - const servicePort = process.env.SERVICE_PORT - ? parseInt(process.env.SERVICE_PORT) - : port; +async function run(options: RunOptions = {}): Promise { + // Check if service is already running + const isRunning = await isServiceRunning(); + if (isRunning) { + console.log("✅ Service is already running in the background."); + return; + } - // Configure logger based on config settings - const pad = num => (num > 9 ? "" : "0") + num; - const generator = (time, index) => { - if (!time) { - time = new Date() - } + // Initialize application + const config = await initializeApp(); + const host = resolveHostConfig(config); + const servicePort = resolveServicePort(config); - var month = time.getFullYear() + "" + pad(time.getMonth() + 1); - var day = pad(time.getDate()); - var hour = pad(time.getHours()); - var minute = pad(time.getMinutes()); + // Save the PID and set up signal handlers + savePid(process.pid); + setupSignalHandlers(); - return `./logs/ccr-${month}${day}${hour}${minute}${pad(time.getSeconds())}${index ? `_${index}` : ''}.log`; - }; - const loggerConfig = - config.LOG !== false - ? { - level: config.LOG_LEVEL || "debug", - stream: createStream(generator, { - path: HOME_DIR, - maxFiles: 3, - interval: "1d", - compress: false, - maxSize: "50M" - }), - } - : false; + // Configure logger + const loggerConfig = configureLogger(config, HOME_DIR); const server = createServer({ jsonPath: CONFIG_FILE, initialConfig: { - // ...config, providers: config.Providers || config.providers, - HOST: HOST, + HOST: host, PORT: servicePort, LOG_FILE: join( homedir(), @@ -138,247 +180,19 @@ async function run(options: RunOptions = {}) { logger: loggerConfig, }); - // Add global error handlers to prevent the service from crashing - process.on("uncaughtException", (err) => { - server.logger.error("Uncaught exception:", err); - }); - - process.on("unhandledRejection", (reason, promise) => { - server.logger.error("Unhandled rejection at:", promise, "reason:", reason); - }); - // Add async preHandler hook for authentication - server.addHook("preHandler", async (req, reply) => { - return new Promise((resolve, reject) => { - const done = (err?: Error) => { - if (err) reject(err); - else resolve(); - }; - // Call the async auth function - apiKeyAuth(config)(req, reply, done).catch(reject); - }); - }); - server.addHook("preHandler", async (req, reply) => { - if (req.url.startsWith("/v1/messages") && !req.url.startsWith("/v1/messages/count_tokens")) { - const useAgents = [] - - for (const agent of agentsManager.getAllAgents()) { - if (agent.shouldHandle(req, config)) { - // 设置agent标识 - useAgents.push(agent.name) - - // change request body - agent.reqHandler(req, config); - - // append agent tools - if (agent.tools.size) { - if (!req.body?.tools?.length) { - req.body.tools = [] - } - req.body.tools.unshift(...Array.from(agent.tools.values()).map(item => { - return { - name: item.name, - description: item.description, - input_schema: item.input_schema - } - })) - } - } - } - - if (useAgents.length) { - req.agents = useAgents; - } - await router(req, reply, { - config, - event - }); - } - }); - server.addHook("onError", async (request, reply, error) => { - event.emit('onError', request, reply, error); - }) - server.addHook("onSend", (req, reply, payload, done) => { - if (req.sessionId && req.url.startsWith("/v1/messages") && !req.url.startsWith("/v1/messages/count_tokens")) { - if (payload instanceof ReadableStream) { - if (req.agents) { - const abortController = new AbortController(); - const eventStream = payload.pipeThrough(new SSEParserTransform()) - let currentAgent: undefined | IAgent; - let currentToolIndex = -1 - let currentToolName = '' - let currentToolArgs = '' - let currentToolId = '' - const toolMessages: any[] = [] - const assistantMessages: any[] = [] - // 存储Anthropic格式的消息体，区分文本和工具类型 - return done(null, rewriteStream(eventStream, async (data, controller) => { - try { - // 检测工具调用开始 - if (data.event === 'content_block_start' && data?.data?.content_block?.name) { - const agent = req.agents.find((name: string) => agentsManager.getAgent(name)?.tools.get(data.data.content_block.name)) - if (agent) { - currentAgent = agentsManager.getAgent(agent) - currentToolIndex = data.data.index - currentToolName = data.data.content_block.name - currentToolId = data.data.content_block.id - return undefined; - } - } - - // 收集工具参数 - if (currentToolIndex > -1 && data.data.index === currentToolIndex && data.data?.delta?.type === 'input_json_delta') { - currentToolArgs += data.data?.delta?.partial_json; - return undefined; - } - - // 工具调用完成，处理agent调用 - if (currentToolIndex > -1 && data.data.index === currentToolIndex && data.data.type === 'content_block_stop') { - try { - const args = JSON5.parse(currentToolArgs); - assistantMessages.push({ - type: "tool_use", - id: currentToolId, - name: currentToolName, - input: args - }) - const toolResult = await currentAgent?.tools.get(currentToolName)?.handler(args, { - req, - config - }); - toolMessages.push({ - "tool_use_id": currentToolId, - "type": "tool_result", - "content": toolResult - }) - currentAgent = undefined - currentToolIndex = -1 - currentToolName = '' - currentToolArgs = '' - currentToolId = '' - } catch (e) { - console.log(e); - } - return undefined; - } - - if (data.event === 'message_delta' && toolMessages.length) { - req.body.messages.push({ - role: 'assistant', - content: assistantMessages - }) - req.body.messages.push({ - role: 'user', - content: toolMessages - }) - const response = await fetch(`http://127.0.0.1:${config.PORT || 3456}/v1/messages`, { - method: "POST", - headers: { - 'x-api-key': config.APIKEY, - 'content-type': 'application/json', - }, - body: JSON.stringify(req.body), - }) - if (!response.ok) { - return undefined; - } - const stream = response.body!.pipeThrough(new SSEParserTransform()) - const reader = stream.getReader() - while (true) { - try { - const {value, done} = await reader.read(); - if (done) { - break; - } - if (['message_start', 'message_stop'].includes(value.event)) { - continue - } - - // 检查流是否仍然可写 - if (!controller.desiredSize) { - break; - } - - controller.enqueue(value) - }catch (readError: any) { - if (readError.name === 'AbortError' || readError.code === 'ERR_STREAM_PREMATURE_CLOSE') { - abortController.abort(); // 中止所有相关操作 - break; - } - throw readError; - } - - } - return undefined - } - return data - }catch (error: any) { - console.error('Unexpected error in stream processing:', error); - - // 处理流提前关闭的错误 - if (error.code === 'ERR_STREAM_PREMATURE_CLOSE') { - abortController.abort(); - return undefined; - } - - // 其他错误仍然抛出 - throw error; - } - }).pipeThrough(new SSESerializerTransform())) - } - - const [originalStream, clonedStream] = payload.tee(); - const read = async (stream: ReadableStream) => { - const reader = stream.getReader(); - try { - while (true) { - const { done, value } = await reader.read(); - if (done) break; - // Process the value if needed - const dataStr = new TextDecoder().decode(value); - if (!dataStr.startsWith("event: message_delta")) { - continue; - } - const str = dataStr.slice(27); - try { - const message = JSON.parse(str); - sessionUsageCache.put(req.sessionId, message.usage); - } catch {} - } - } catch (readError: any) { - if (readError.name === 'AbortError' || readError.code === 'ERR_STREAM_PREMATURE_CLOSE') { - console.error('Background read stream closed prematurely'); - } else { - console.error('Error in background stream reading:', readError); - } - } finally { - reader.releaseLock(); - } - } - read(clonedStream); - return done(null, originalStream) - } - sessionUsageCache.put(req.sessionId, payload.usage); - if (typeof payload ==='object') { - if (payload.error) { - return done(payload.error, null) - } else { - return done(payload, null) - } - } - } - if (typeof payload ==='object' && payload.error) { - return done(payload.error, null) - } - done(null, payload) - }); - server.addHook("onSend", async (req, reply, payload) => { - event.emit('onSend', req, reply, payload); - return payload; - }) - + // Setup global error handlers and hooks + setupGlobalErrorHandlers(server); + setupRequestLoggingHook(server); + setupResponseLoggingHook(server); + setupAuthHook(server, config); + setupAgentAndRoutingHook(server, config, event); + setupErrorEventHook(server, event); + setupSendEventHook(server, event); + setupAgentProcessingHook(server, config); + setupSessionUsageHook(server, config); + setupErrorPayloadHook(server); server.start(); } export { run }; -// run(); diff --git a/src/utils/hooks.ts b/src/utils/hooks.ts new file mode 100644 index 00000000..9828db39 --- /dev/null +++ b/src/utils/hooks.ts @@ -0,0 +1,552 @@ +import { EventEmitter } from "node:events"; +import { IAgent } from "../agents/type"; +import agentsManager from "../agents"; +import { sessionUsageCache } from "./cache"; +import { SSEParserTransform } from "./SSEParser.transform"; +import { SSESerializerTransform } from "./SSESerializer.transform"; +import { rewriteStream } from "./rewriteStream"; +import JSON5 from "json5"; +import { apiKeyAuth } from "../middleware/auth"; +import { router } from "./router"; + +/** + * Setup request logging hook + */ +export function setupRequestLoggingHook(server: any): void { + server.addHook("preHandler", async (req, reply) => { + const requestData = { + method: req.method, + url: req.url, + headers: req.headers, + body: req.body, + query: req.query, + timestamp: new Date().toISOString() + }; + + req.log.info({ + requestData, + msg: "Request details" + }, "Incoming request"); + }); +} + +/** + * Setup response logging hook + */ +export function setupResponseLoggingHook(server: any): void { + server.addHook("onSend", async (req, reply, payload) => { + const responseData = { + method: req.method, + url: req.url, + statusCode: reply.statusCode, + headers: reply.getHeaders(), + timestamp: new Date().toISOString() + }; + + let responseBody: any; + + if (payload instanceof ReadableStream) { + responseBody = { + type: "ReadableStream", + readable: true, + note: "Streaming response - complete content will be logged when stream ends" + }; + + payload = createLoggingWrappedStream(payload, req, reply); + } else if (payload === null || payload === undefined) { + responseBody = null; + } else if (typeof payload === 'string') { + responseBody = payload; + } else if (Buffer.isBuffer(payload)) { + responseBody = payload.toString('utf8'); + } else if (typeof payload === 'object') { + responseBody = payload; + } else { + responseBody = { type: typeof payload, content: payload }; + } + + responseData.body = responseBody; + + req.log.info({ + responseData, + msg: "Response details" + }, "Response completed"); + + return payload; + }); +} + +/** + * Setup authentication hook + */ +export function setupAuthHook(server: any, config: any): void { + server.addHook("preHandler", async (req, reply) => { + return new Promise((resolve, reject) => { + const done = (err?: Error) => { + if (err) reject(err); + else resolve(); + }; + apiKeyAuth(config)(req, reply, done).catch(reject); + }); + }); +} + +/** + * Setup agent and routing hook + */ +export function setupAgentAndRoutingHook(server: any, config: any, event: EventEmitter): void { + server.addHook("preHandler", async (req, reply) => { + if (req.url.startsWith("/v1/messages") && !req.url.startsWith("/v1/messages/count_tokens")) { + const activeAgents: string[] = []; + + for (const agent of agentsManager.getAllAgents()) { + if (agent.shouldHandle(req, config)) { + activeAgents.push(agent.name); + agent.reqHandler(req, config); + + if (agent.tools.size) { + if (!req.body?.tools?.length) { + req.body.tools = []; + } + req.body.tools.unshift(...Array.from(agent.tools.values()).map(item => ({ + name: item.name, + description: item.description, + input_schema: item.input_schema + }))); + } + } + } + + if (activeAgents.length) { + req.agents = activeAgents; + } + + await router(req, reply, { config, event }); + } + }); +} + +/** + * Setup error event hook + */ +export function setupErrorEventHook(server: any, event: EventEmitter): void { + server.addHook("onError", async (request, reply, error) => { + event.emit('onError', request, reply, error); + }); +} + +/** + * Setup send event hook + */ +export function setupSendEventHook(server: any, event: EventEmitter): void { + server.addHook("onSend", async (req, reply, payload) => { + event.emit('onSend', req, reply, payload); + return payload; + }); +} + +/** + * Setup agent processing hook for handling agent interactions + */ +export function setupAgentProcessingHook(server: any, config: any): void { + server.addHook("onSend", async (req, reply, payload) => { + if (req.sessionId && req.url.startsWith("/v1/messages") && !req.url.startsWith("/v1/messages/count_tokens")) { + if (payload instanceof ReadableStream) { + if (req.agents) { + return handleAgentStreamProcessing(payload, req, config); + } + + const [originalStream, clonedStream] = payload.tee(); + readSessionUsageFromStream(clonedStream, req.sessionId); + return originalStream; + } + + sessionUsageCache.put(req.sessionId, payload.usage); + if (typeof payload === 'object') { + if (payload.error) { + throw payload.error; + } + return payload; + } + } + + if (typeof payload === 'object' && payload.error) { + throw payload.error; + } + + return payload; + }); +} + +/** + * Setup session usage hook + */ +export function setupSessionUsageHook(server: any, config: any): void { + // Session usage tracking is now handled in setupAgentProcessingHook +} + +/** + * Setup error payload hook + */ +export function setupErrorPayloadHook(server: any): void { + // Error payload handling is now integrated in setupAgentProcessingHook +} + +/** + * Pads a number with leading zero if needed + */ +function padZero(num: number): string { + return (num > 9 ? "" : "0") + num; +} + +/** + * Generates log file names with timestamp + */ +function logFileNameGenerator(time?: Date, index?: number): string { + if (!time) { + time = new Date(); + } + + const yearAndMonth = time.getFullYear() + "" + padZero(time.getMonth() + 1); + const day = padZero(time.getDate()); + const hour = padZero(time.getHours()); + const minute = padZero(time.getMinutes()); + const second = padZero(time.getSeconds()); + + return `./logs/ccr-${yearAndMonth}${day}${hour}${minute}${second}${index ? `_${index}` : ''}.log`; +} + +/** + * Parse SSE content to extract structured information + */ +function parseSSEContent(fullContent: string): any { + try { + const events = fullContent.split('\n\n').filter(event => event.trim()); + const messages = []; + + for (const event of events) { + const lines = event.split('\n'); + for (const line of lines) { + if (line.startsWith('data: ') && line.length > 6) { + try { + const data = JSON.parse(line.substring(6)); + if (data.type === 'message_delta' && data.usage) { + messages.push({ type: 'usage', data: data.usage }); + } else if (data.type === 'content_block_delta' && data.delta?.text) { + messages.push({ type: 'text', data: data.delta.text }); + } else if (data.type === 'content_block_start' && data.content_block?.type === 'text') { + messages.push({ type: 'text_start' }); + } + } catch (e) { + // Ignore parsing errors and continue processing other data + } + } + } + } + + if (messages.length > 0) { + return { + type: "parsed_sse_response", + events: messages, + fullSSEContent: fullContent, + summary: { + totalEvents: events.length, + hasUsage: messages.some(m => m.type === 'usage'), + textEvents: messages.filter(m => m.type === 'text').length + } + }; + } + } catch (parseError) { + console.warn('Failed to parse SSE content:', parseError.message); + } + + return fullContent; +} + +/** + * Create a wrapped stream that logs complete response content + */ +function createLoggingWrappedStream( + originalStream: ReadableStream, + req: any, + reply: any +): ReadableStream { + const loggedChunks: string[] = []; + const startTime = Date.now(); + + return new ReadableStream({ + async start(controller) { + const reader = originalStream.getReader(); + + try { + while (true) { + const { done, value } = await reader.read(); + + if (done) { + const fullContent = loggedChunks.join(''); + const endTime = Date.now(); + const duration = endTime - startTime; + const parsedContent = parseSSEContent(fullContent); + + req.log.info({ + streamCompleteResponse: { + type: "ReadableStream_complete_response", + content: parsedContent, + metadata: { + totalChunks: loggedChunks.length, + contentSize: fullContent.length, + duration: duration, + method: req.method, + url: req.url, + statusCode: reply.statusCode, + timestamp: new Date().toISOString() + } + }, + msg: "Complete stream response" + }, "Stream response completed - full body"); + + controller.close(); + break; + } + + const chunkText = new TextDecoder().decode(value); + loggedChunks.push(chunkText); + controller.enqueue(value); + } + } catch (error) { + const endTime = Date.now(); + const duration = endTime - startTime; + + req.log.error({ + streamError: { + error: error.message, + stack: error.stack, + metadata: { + chunksCollected: loggedChunks.length, + duration: duration, + method: req.method, + url: req.url, + timestamp: new Date().toISOString() + } + } + }, "Stream reading error"); + + controller.error(error); + } + } + }); +} + +/** + * Handle agent stream processing with tool calls + */ +function handleAgentStreamProcessing(payload: ReadableStream, req: any, config: any): ReadableStream { + const abortController = new AbortController(); + const eventStream = payload.pipeThrough(new SSEParserTransform()); + + const agentState = { + currentAgent: undefined as IAgent | undefined, + currentToolIndex: -1, + currentToolName: '', + currentToolArgs: '', + currentToolId: '', + toolMessages: [] as any[], + assistantMessages: [] as any[] + }; + + return rewriteStream(eventStream, async (data, controller) => { + try { + // Handle tool call start + if (data.event === 'content_block_start' && data?.data?.content_block?.name) { + const agent = req.agents.find((name: string) => + agentsManager.getAgent(name)?.tools.get(data.data.content_block.name) + ); + + if (agent) { + agentState.currentAgent = agentsManager.getAgent(agent); + agentState.currentToolIndex = data.data.index; + agentState.currentToolName = data.data.content_block.name; + agentState.currentToolId = data.data.content_block.id; + return undefined; + } + } + + // Collect tool arguments + if (agentState.currentToolIndex > -1 && + data.data.index === agentState.currentToolIndex && + data.data?.delta?.type === 'input_json_delta') { + agentState.currentToolArgs += data.data?.delta?.partial_json; + return undefined; + } + + // Handle tool call completion + if (agentState.currentToolIndex > -1 && + data.data.index === agentState.currentToolIndex && + data.data.type === 'content_block_stop') { + await processToolCall(agentState, req, config); + return undefined; + } + + // Handle message delta with tool results + if (data.event === 'message_delta' && agentState.toolMessages.length) { + return await retryRequestWithToolResults(agentState, req, config, controller, abortController); + } + + return data; + } catch (error: any) { + handleStreamProcessingError(error, abortController); + } + }).pipeThrough(new SSESerializerTransform()); +} + +/** + * Process a completed tool call + */ +async function processToolCall(agentState: any, req: any, config: any): Promise { + try { + const args = JSON5.parse(agentState.currentToolArgs); + + agentState.assistantMessages.push({ + type: "tool_use", + id: agentState.currentToolId, + name: agentState.currentToolName, + input: args + }); + + const toolResult = await agentState.currentAgent?.tools.get(agentState.currentToolName)?.handler(args, { + req, + config + }); + + agentState.toolMessages.push({ + "tool_use_id": agentState.currentToolId, + "type": "tool_result", + "content": toolResult + }); + + // Reset agent state + agentState.currentAgent = undefined; + agentState.currentToolIndex = -1; + agentState.currentToolName = ''; + agentState.currentToolArgs = ''; + agentState.currentToolId = ''; + } catch (e) { + console.error('Error processing tool call:', e); + } +} + +/** + * Retry request with tool results + */ +async function retryRequestWithToolResults( + agentState: any, + req: any, + config: any, + controller: any, + abortController: AbortController +): Promise { + req.body.messages.push({ + role: 'assistant', + content: agentState.assistantMessages + }); + + req.body.messages.push({ + role: 'user', + content: agentState.toolMessages + }); + + const response = await fetch(`http://127.0.0.1:${config.PORT || 3456}/v1/messages`, { + method: "POST", + headers: { + 'x-api-key': config.APIKEY, + 'content-type': 'application/json', + }, + body: JSON.stringify(req.body), + }); + + if (!response.ok) { + return undefined; + } + + const stream = response.body!.pipeThrough(new SSEParserTransform()); + const reader = stream.getReader(); + + while (true) { + try { + const { value, done } = await reader.read(); + if (done) { + break; + } + + if (['message_start', 'message_stop'].includes(value.event)) { + continue; + } + + if (!controller.desiredSize) { + break; + } + + controller.enqueue(value); + } catch (readError: any) { + if (readError.name === 'AbortError' || readError.code === 'ERR_STREAM_PREMATURE_CLOSE') { + abortController.abort(); + break; + } + throw readError; + } + } + + return undefined; +} + +/** + * Handle stream processing errors + */ +function handleStreamProcessingError(error: any, abortController: AbortController): void { + console.error('Unexpected error in stream processing:', error); + + if (error.code === 'ERR_STREAM_PREMATURE_CLOSE') { + abortController.abort(); + return undefined; + } + + throw error; +} + +/** + * Read session usage from stream in background + */ +function readSessionUsageFromStream(stream: ReadableStream, sessionId: string): void { + const read = async (stream: ReadableStream) => { + const reader = stream.getReader(); + try { + while (true) { + const { done, value } = await reader.read(); + if (done) break; + + const dataStr = new TextDecoder().decode(value); + if (!dataStr.startsWith("event: message_delta")) { + continue; + } + + const str = dataStr.slice(27); + try { + const message = JSON.parse(str); + sessionUsageCache.put(sessionId, message.usage); + } catch { + // Ignore parsing errors + } + } + } catch (readError: any) { + if (readError.name === 'AbortError' || readError.code === 'ERR_STREAM_PREMATURE_CLOSE') { + console.error('Background read stream closed prematurely'); + } else { + console.error('Error in background stream reading:', readError); + } + } finally { + reader.releaseLock(); + } + }; + + read(stream); +} \ No newline at end of file From a22fc837e9cdcd9a03c50b78ffe2efed225999a5 Mon Sep 17 00:00:00 2001 From: Fu Gui Date: Tue, 25 Nov 2025 17:30:18 +0800 Subject: [PATCH 2/6] refactor: extract stream processing and agent utilities into separate modules MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Split complex stream processing logic from hooks.ts into dedicated modules - Add streamUtils.ts for SSE parsing and stream handling utilities - Add agentProcessor.ts for agent-specific stream processing logic - Add types.ts for shared type definitions - Improve code organization and maintainability of hooks system 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- .claude/settings.local.json | 17 ++ src/utils/agentProcessor.ts | 349 ++++++++++++++++++++++++ src/utils/hooks.ts | 510 ++++++++---------------------------- src/utils/streamUtils.ts | 234 +++++++++++++++++ src/utils/types.ts | 73 ++++++ 5 files changed, 786 insertions(+), 397 deletions(-) create mode 100644 .claude/settings.local.json create mode 100644 src/utils/agentProcessor.ts create mode 100644 src/utils/streamUtils.ts create mode 100644 src/utils/types.ts diff --git a/.claude/settings.local.json b/.claude/settings.local.json new file mode 100644 index 00000000..5f67ac46 --- /dev/null +++ b/.claude/settings.local.json @@ -0,0 +1,17 @@ +{ + "permissions": { + "allow": [ + "Bash(dir:*)", + "Bash(npm run build)", + "Bash(ccr restart:*)", + "Bash(ccr:*)", + "Bash(curl:*)", + "Bash(Select-Object -Last 30)", + "Bash(git log:*)", + "Bash(git add:*)", + "Bash(git commit:*)" + ], + "deny": [], + "ask": [] + } +} diff --git a/src/utils/agentProcessor.ts b/src/utils/agentProcessor.ts new file mode 100644 index 00000000..b385e1f8 --- /dev/null +++ b/src/utils/agentProcessor.ts @@ -0,0 +1,349 @@ +import agentsManager from "../agents"; +import { SSEParserTransform } from "./SSEParser.transform"; +import { SSESerializerTransform } from "./SSESerializer.transform"; +import { rewriteStream } from "./rewriteStream"; +import JSON5 from "json5"; +import { + AgentProcessingState, + AssistantMessage, + ToolMessage +} from "./types"; + +/** + * Handle agent stream processing with tool calls + */ +export function handleAgentStreamProcessing( + payload: ReadableStream, + req: any, + config: any +): ReadableStream { + const abortController = new AbortController(); + const eventStream = payload.pipeThrough(new SSEParserTransform()); + const agentState = createInitialAgentState(); + + return rewriteStream(eventStream, async (data, controller) => { + try { + const result = await processStreamEvent(data, agentState, req, config); + + if (result === 'continue') { + return undefined; + } + + if (result === 'retry_with_tools') { + await retryRequestWithToolResults(agentState, req, config, controller, abortController); + return undefined; + } + + return data; + } catch (error: any) { + handleStreamProcessingError(error, abortController); + } + }).pipeThrough(new SSESerializerTransform()); +} + +/** + * Create initial agent processing state + */ +function createInitialAgentState(): AgentProcessingState { + return { + currentAgent: undefined, + currentToolIndex: -1, + currentToolName: '', + currentToolArgs: '', + currentToolId: '', + toolMessages: [], + assistantMessages: [] + }; +} + +/** + * Process individual stream events + */ +async function processStreamEvent( + data: any, + agentState: AgentProcessingState, + req: any, + config: any +): Promise<'continue' | 'retry_with_tools' | 'pass_through'> { + // Handle tool call start + if (isToolCallStart(data)) { + const agent = findAgentForTool(data.data.content_block.name, req); + if (agent) { + updateAgentStateForToolStart(agentState, agent, data); + return 'continue'; + } + } + + // Collect tool arguments + if (isToolArgumentCollection(data, agentState)) { + collectToolArguments(agentState, data); + return 'continue'; + } + + // Handle tool call completion + if (isToolCallCompletion(data, agentState)) { + await processToolCall(agentState, req, config); + return 'continue'; + } + + // Handle message delta with tool results + if (isMessageDeltaWithToolResults(data, agentState)) { + return 'retry_with_tools'; + } + + return 'pass_through'; +} + +/** + * Check if event is tool call start + */ +function isToolCallStart(data: any): boolean { + return data.event === 'content_block_start' && + data?.data?.content_block?.name; +} + +/** + * Find agent that can handle the tool + */ +function findAgentForTool(toolName: string, req: any): string | undefined { + return req.agents.find((name: string) => + agentsManager.getAgent(name)?.tools.get(toolName) + ); +} + +/** + * Update agent state when tool call starts + */ +function updateAgentStateForToolStart( + agentState: AgentProcessingState, + agentName: string, + data: any +): void { + const agent = agentsManager.getAgent(agentName); + agentState.currentAgent = agent; + agentState.currentToolIndex = data.data.index; + agentState.currentToolName = data.data.content_block.name; + agentState.currentToolId = data.data.content_block.id; +} + +/** + * Check if event is for collecting tool arguments + */ +function isToolArgumentCollection(data: any, agentState: AgentProcessingState): boolean { + return agentState.currentToolIndex > -1 && + data.data.index === agentState.currentToolIndex && + data.data?.delta?.type === 'input_json_delta'; +} + +/** + * Collect tool arguments from delta + */ +function collectToolArguments(agentState: AgentProcessingState, data: any): void { + agentState.currentToolArgs += data.data?.delta?.partial_json; +} + +/** + * Check if event is tool call completion + */ +function isToolCallCompletion(data: any, agentState: AgentProcessingState): boolean { + return agentState.currentToolIndex > -1 && + data.data.index === agentState.currentToolIndex && + data.data.type === 'content_block_stop'; +} + +/** + * Process a completed tool call + */ +async function processToolCall( + agentState: AgentProcessingState, + req: any, + config: any +): Promise { + try { + const args = JSON5.parse(agentState.currentToolArgs); + + const assistantMessage = createAssistantMessage(agentState); + agentState.assistantMessages.push(assistantMessage); + + const toolResult = await executeTool(agentState, args, req, config); + const toolMessage = createToolMessage(agentState.currentToolId, toolResult); + agentState.toolMessages.push(toolMessage); + + resetAgentToolState(agentState); + } catch (error) { + console.error('Error processing tool call:', error); + } +} + +/** + * Create assistant message for tool use + */ +function createAssistantMessage(agentState: AgentProcessingState): AssistantMessage { + return { + type: "tool_use", + id: agentState.currentToolId, + name: agentState.currentToolName, + input: JSON5.parse(agentState.currentToolArgs) + }; +} + +/** + * Execute tool and get result + */ +async function executeTool( + agentState: AgentProcessingState, + args: any, + req: any, + config: any +): Promise { + const tool = agentState.currentAgent?.tools.get(agentState.currentToolName); + if (!tool) { + throw new Error(`Tool ${agentState.currentToolName} not found`); + } + + return await tool.handler(args, { req, config }); +} + +/** + * Create tool result message + */ +function createToolMessage(toolUseId: string, toolResult: string): ToolMessage { + return { + tool_use_id: toolUseId, + type: "tool_result", + content: toolResult + }; +} + +/** + * Reset agent tool-specific state + */ +function resetAgentToolState(agentState: AgentProcessingState): void { + agentState.currentAgent = undefined; + agentState.currentToolIndex = -1; + agentState.currentToolName = ''; + agentState.currentToolArgs = ''; + agentState.currentToolId = ''; +} + +/** + * Check if event is message delta with tool results + */ +function isMessageDeltaWithToolResults(data: any, agentState: AgentProcessingState): boolean { + return data.event === 'message_delta' && agentState.toolMessages.length > 0; +} + +/** + * Retry request with tool results + */ +async function retryRequestWithToolResults( + agentState: AgentProcessingState, + req: any, + config: any, + controller: any, + abortController: AbortController +): Promise { + addToolMessagesToRequest(req, agentState); + + const response = await makeRetryRequest(req, config); + if (!response.ok) { + return; + } + + await processRetryResponse(response, controller, abortController); +} + +/** + * Add tool messages to request body + */ +function addToolMessagesToRequest(req: any, agentState: AgentProcessingState): void { + req.body.messages.push({ + role: 'assistant', + content: agentState.assistantMessages + }); + + req.body.messages.push({ + role: 'user', + content: agentState.toolMessages + }); +} + +/** + * Make retry request with tool results + */ +async function makeRetryRequest(req: any, config: any): Promise { + return await fetch(`http://127.0.0.1:${config.PORT || 3456}/v1/messages`, { + method: "POST", + headers: { + 'x-api-key': config.APIKEY, + 'content-type': 'application/json', + }, + body: JSON.stringify(req.body), + }); +} + +/** + * Process retry response stream + */ +async function processRetryResponse( + response: Response, + controller: any, + abortController: AbortController +): Promise { + const stream = response.body!.pipeThrough(new SSEParserTransform()); + const reader = stream.getReader(); + + while (true) { + try { + const { value, done } = await reader.read(); + if (done) { + break; + } + + if (shouldSkipEvent(value)) { + continue; + } + + if (!controller.desiredSize) { + break; + } + + controller.enqueue(value); + } catch (readError: any) { + handleRetryStreamError(readError, abortController); + break; + } + } +} + +/** + * Check if event should be skipped + */ +function shouldSkipEvent(value: any): boolean { + return ['message_start', 'message_stop'].includes(value.event); +} + +/** + * Handle retry stream errors + */ +function handleRetryStreamError(readError: any, abortController: AbortController): void { + if (readError.name === 'AbortError' || readError.code === 'ERR_STREAM_PREMATURE_CLOSE') { + abortController.abort(); + } else { + throw readError; + } +} + +/** + * Handle stream processing errors + */ +function handleStreamProcessingError(error: any, abortController: AbortController): void { + console.error('Unexpected error in stream processing:', error); + + if (error.code === 'ERR_STREAM_PREMATURE_CLOSE') { + abortController.abort(); + return; + } + + throw error; +} \ No newline at end of file diff --git a/src/utils/hooks.ts b/src/utils/hooks.ts index 9828db39..219e9be0 100644 --- a/src/utils/hooks.ts +++ b/src/utils/hooks.ts @@ -2,12 +2,14 @@ import { EventEmitter } from "node:events"; import { IAgent } from "../agents/type"; import agentsManager from "../agents"; import { sessionUsageCache } from "./cache"; -import { SSEParserTransform } from "./SSEParser.transform"; -import { SSESerializerTransform } from "./SSESerializer.transform"; -import { rewriteStream } from "./rewriteStream"; -import JSON5 from "json5"; import { apiKeyAuth } from "../middleware/auth"; import { router } from "./router"; +import { + createLoggingWrappedStream, + readSessionUsageFromStream, + parseSSEContent +} from "./streamUtils"; +import { handleAgentStreamProcessing } from "./agentProcessor"; /** * Setup request logging hook @@ -43,30 +45,19 @@ export function setupResponseLoggingHook(server: any): void { timestamp: new Date().toISOString() }; - let responseBody: any; + const responseBody = extractResponseBody(payload); if (payload instanceof ReadableStream) { - responseBody = { + responseData.body = { type: "ReadableStream", readable: true, note: "Streaming response - complete content will be logged when stream ends" }; - payload = createLoggingWrappedStream(payload, req, reply); - } else if (payload === null || payload === undefined) { - responseBody = null; - } else if (typeof payload === 'string') { - responseBody = payload; - } else if (Buffer.isBuffer(payload)) { - responseBody = payload.toString('utf8'); - } else if (typeof payload === 'object') { - responseBody = payload; } else { - responseBody = { type: typeof payload, content: payload }; + responseData.body = responseBody; } - responseData.body = responseBody; - req.log.info({ responseData, msg: "Response details" @@ -76,6 +67,29 @@ export function setupResponseLoggingHook(server: any): void { }); } +/** + * Extract response body for logging + */ +function extractResponseBody(payload: any): any { + if (payload === null || payload === undefined) { + return null; + } + + if (typeof payload === 'string') { + return payload; + } + + if (Buffer.isBuffer(payload)) { + return payload.toString('utf8'); + } + + if (typeof payload === 'object') { + return payload; + } + + return { type: typeof payload, content: payload }; +} + /** * Setup authentication hook */ @@ -96,26 +110,8 @@ export function setupAuthHook(server: any, config: any): void { */ export function setupAgentAndRoutingHook(server: any, config: any, event: EventEmitter): void { server.addHook("preHandler", async (req, reply) => { - if (req.url.startsWith("/v1/messages") && !req.url.startsWith("/v1/messages/count_tokens")) { - const activeAgents: string[] = []; - - for (const agent of agentsManager.getAllAgents()) { - if (agent.shouldHandle(req, config)) { - activeAgents.push(agent.name); - agent.reqHandler(req, config); - - if (agent.tools.size) { - if (!req.body?.tools?.length) { - req.body.tools = []; - } - req.body.tools.unshift(...Array.from(agent.tools.values()).map(item => ({ - name: item.name, - description: item.description, - input_schema: item.input_schema - }))); - } - } - } + if (isMessagesEndpoint(req)) { + const activeAgents = processActiveAgents(req, config); if (activeAgents.length) { req.agents = activeAgents; @@ -126,6 +122,47 @@ export function setupAgentAndRoutingHook(server: any, config: any, event: EventE }); } +/** + * Check if request is for messages endpoint + */ +function isMessagesEndpoint(req: any): boolean { + return req.url.startsWith("/v1/messages") && !req.url.startsWith("/v1/messages/count_tokens"); +} + +/** + * Process active agents and update request + */ +function processActiveAgents(req: any, config: any): string[] { + const activeAgents: string[] = []; + + for (const agent of agentsManager.getAllAgents()) { + if (agent.shouldHandle(req, config)) { + activeAgents.push(agent.name); + agent.reqHandler(req, config); + addAgentToolsToRequest(agent, req); + } + } + + return activeAgents; +} + +/** + * Add agent tools to request body + */ +function addAgentToolsToRequest(agent: IAgent, req: any): void { + if (agent.tools.size) { + if (!req.body?.tools?.length) { + req.body.tools = []; + } + const tools = Array.from(agent.tools.values()).map(item => ({ + name: item.name, + description: item.description, + input_schema: item.input_schema + })); + req.body.tools.unshift(...tools); + } +} + /** * Setup error event hook */ @@ -150,403 +187,82 @@ export function setupSendEventHook(server: any, event: EventEmitter): void { */ export function setupAgentProcessingHook(server: any, config: any): void { server.addHook("onSend", async (req, reply, payload) => { - if (req.sessionId && req.url.startsWith("/v1/messages") && !req.url.startsWith("/v1/messages/count_tokens")) { - if (payload instanceof ReadableStream) { - if (req.agents) { - return handleAgentStreamProcessing(payload, req, config); - } - - const [originalStream, clonedStream] = payload.tee(); - readSessionUsageFromStream(clonedStream, req.sessionId); - return originalStream; - } - - sessionUsageCache.put(req.sessionId, payload.usage); - if (typeof payload === 'object') { - if (payload.error) { - throw payload.error; - } - return payload; - } + if (shouldProcessAgentRequest(req)) { + return await processAgentRequest(req, payload, config); } - if (typeof payload === 'object' && payload.error) { - throw payload.error; - } - - return payload; + return handleErrorResponse(payload); }); } /** - * Setup session usage hook + * Check if request should be processed by agent */ -export function setupSessionUsageHook(server: any, config: any): void { - // Session usage tracking is now handled in setupAgentProcessingHook +function shouldProcessAgentRequest(req: any): boolean { + return req.sessionId && isMessagesEndpoint(req); } /** - * Setup error payload hook + * Process agent request */ -export function setupErrorPayloadHook(server: any): void { - // Error payload handling is now integrated in setupAgentProcessingHook -} - -/** - * Pads a number with leading zero if needed - */ -function padZero(num: number): string { - return (num > 9 ? "" : "0") + num; -} - -/** - * Generates log file names with timestamp - */ -function logFileNameGenerator(time?: Date, index?: number): string { - if (!time) { - time = new Date(); +async function processAgentRequest(req: any, payload: any, config: any): Promise { + if (payload instanceof ReadableStream) { + return handleStreamPayload(payload, req, config); } - const yearAndMonth = time.getFullYear() + "" + padZero(time.getMonth() + 1); - const day = padZero(time.getDate()); - const hour = padZero(time.getHours()); - const minute = padZero(time.getMinutes()); - const second = padZero(time.getSeconds()); - - return `./logs/ccr-${yearAndMonth}${day}${hour}${minute}${second}${index ? `_${index}` : ''}.log`; + return handleNonStreamPayload(payload, req); } /** - * Parse SSE content to extract structured information + * Handle stream payload */ -function parseSSEContent(fullContent: string): any { - try { - const events = fullContent.split('\n\n').filter(event => event.trim()); - const messages = []; - - for (const event of events) { - const lines = event.split('\n'); - for (const line of lines) { - if (line.startsWith('data: ') && line.length > 6) { - try { - const data = JSON.parse(line.substring(6)); - if (data.type === 'message_delta' && data.usage) { - messages.push({ type: 'usage', data: data.usage }); - } else if (data.type === 'content_block_delta' && data.delta?.text) { - messages.push({ type: 'text', data: data.delta.text }); - } else if (data.type === 'content_block_start' && data.content_block?.type === 'text') { - messages.push({ type: 'text_start' }); - } - } catch (e) { - // Ignore parsing errors and continue processing other data - } - } - } - } - - if (messages.length > 0) { - return { - type: "parsed_sse_response", - events: messages, - fullSSEContent: fullContent, - summary: { - totalEvents: events.length, - hasUsage: messages.some(m => m.type === 'usage'), - textEvents: messages.filter(m => m.type === 'text').length - } - }; - } - } catch (parseError) { - console.warn('Failed to parse SSE content:', parseError.message); +function handleStreamPayload(payload: ReadableStream, req: any, config: any): ReadableStream { + if (req.agents) { + return handleAgentStreamProcessing(payload, req, config); } - return fullContent; + const [originalStream, clonedStream] = payload.tee(); + readSessionUsageFromStream(clonedStream, req.sessionId); + return originalStream; } /** - * Create a wrapped stream that logs complete response content + * Handle non-stream payload */ -function createLoggingWrappedStream( - originalStream: ReadableStream, - req: any, - reply: any -): ReadableStream { - const loggedChunks: string[] = []; - const startTime = Date.now(); - - return new ReadableStream({ - async start(controller) { - const reader = originalStream.getReader(); - - try { - while (true) { - const { done, value } = await reader.read(); - - if (done) { - const fullContent = loggedChunks.join(''); - const endTime = Date.now(); - const duration = endTime - startTime; - const parsedContent = parseSSEContent(fullContent); - - req.log.info({ - streamCompleteResponse: { - type: "ReadableStream_complete_response", - content: parsedContent, - metadata: { - totalChunks: loggedChunks.length, - contentSize: fullContent.length, - duration: duration, - method: req.method, - url: req.url, - statusCode: reply.statusCode, - timestamp: new Date().toISOString() - } - }, - msg: "Complete stream response" - }, "Stream response completed - full body"); - - controller.close(); - break; - } - - const chunkText = new TextDecoder().decode(value); - loggedChunks.push(chunkText); - controller.enqueue(value); - } - } catch (error) { - const endTime = Date.now(); - const duration = endTime - startTime; - - req.log.error({ - streamError: { - error: error.message, - stack: error.stack, - metadata: { - chunksCollected: loggedChunks.length, - duration: duration, - method: req.method, - url: req.url, - timestamp: new Date().toISOString() - } - } - }, "Stream reading error"); - - controller.error(error); - } - } - }); -} +function handleNonStreamPayload(payload: any, req: any): any { + sessionUsageCache.put(req.sessionId, payload.usage); -/** - * Handle agent stream processing with tool calls - */ -function handleAgentStreamProcessing(payload: ReadableStream, req: any, config: any): ReadableStream { - const abortController = new AbortController(); - const eventStream = payload.pipeThrough(new SSEParserTransform()); - - const agentState = { - currentAgent: undefined as IAgent | undefined, - currentToolIndex: -1, - currentToolName: '', - currentToolArgs: '', - currentToolId: '', - toolMessages: [] as any[], - assistantMessages: [] as any[] - }; - - return rewriteStream(eventStream, async (data, controller) => { - try { - // Handle tool call start - if (data.event === 'content_block_start' && data?.data?.content_block?.name) { - const agent = req.agents.find((name: string) => - agentsManager.getAgent(name)?.tools.get(data.data.content_block.name) - ); - - if (agent) { - agentState.currentAgent = agentsManager.getAgent(agent); - agentState.currentToolIndex = data.data.index; - agentState.currentToolName = data.data.content_block.name; - agentState.currentToolId = data.data.content_block.id; - return undefined; - } - } - - // Collect tool arguments - if (agentState.currentToolIndex > -1 && - data.data.index === agentState.currentToolIndex && - data.data?.delta?.type === 'input_json_delta') { - agentState.currentToolArgs += data.data?.delta?.partial_json; - return undefined; - } - - // Handle tool call completion - if (agentState.currentToolIndex > -1 && - data.data.index === agentState.currentToolIndex && - data.data.type === 'content_block_stop') { - await processToolCall(agentState, req, config); - return undefined; - } - - // Handle message delta with tool results - if (data.event === 'message_delta' && agentState.toolMessages.length) { - return await retryRequestWithToolResults(agentState, req, config, controller, abortController); - } - - return data; - } catch (error: any) { - handleStreamProcessingError(error, abortController); + if (typeof payload === 'object') { + if (payload.error) { + throw payload.error; } - }).pipeThrough(new SSESerializerTransform()); -} - -/** - * Process a completed tool call - */ -async function processToolCall(agentState: any, req: any, config: any): Promise { - try { - const args = JSON5.parse(agentState.currentToolArgs); - - agentState.assistantMessages.push({ - type: "tool_use", - id: agentState.currentToolId, - name: agentState.currentToolName, - input: args - }); - - const toolResult = await agentState.currentAgent?.tools.get(agentState.currentToolName)?.handler(args, { - req, - config - }); - - agentState.toolMessages.push({ - "tool_use_id": agentState.currentToolId, - "type": "tool_result", - "content": toolResult - }); - - // Reset agent state - agentState.currentAgent = undefined; - agentState.currentToolIndex = -1; - agentState.currentToolName = ''; - agentState.currentToolArgs = ''; - agentState.currentToolId = ''; - } catch (e) { - console.error('Error processing tool call:', e); + return payload; } } /** - * Retry request with tool results + * Handle error payload */ -async function retryRequestWithToolResults( - agentState: any, - req: any, - config: any, - controller: any, - abortController: AbortController -): Promise { - req.body.messages.push({ - role: 'assistant', - content: agentState.assistantMessages - }); - - req.body.messages.push({ - role: 'user', - content: agentState.toolMessages - }); - - const response = await fetch(`http://127.0.0.1:${config.PORT || 3456}/v1/messages`, { - method: "POST", - headers: { - 'x-api-key': config.APIKEY, - 'content-type': 'application/json', - }, - body: JSON.stringify(req.body), - }); - - if (!response.ok) { - return undefined; +function handleErrorResponse(payload: any): any { + if (typeof payload === 'object' && payload.error) { + throw payload.error; } - const stream = response.body!.pipeThrough(new SSEParserTransform()); - const reader = stream.getReader(); - - while (true) { - try { - const { value, done } = await reader.read(); - if (done) { - break; - } - - if (['message_start', 'message_stop'].includes(value.event)) { - continue; - } - - if (!controller.desiredSize) { - break; - } - - controller.enqueue(value); - } catch (readError: any) { - if (readError.name === 'AbortError' || readError.code === 'ERR_STREAM_PREMATURE_CLOSE') { - abortController.abort(); - break; - } - throw readError; - } - } - - return undefined; + return payload; } /** - * Handle stream processing errors + * Setup session usage hook + * @deprecated This functionality is now integrated in setupAgentProcessingHook */ -function handleStreamProcessingError(error: any, abortController: AbortController): void { - console.error('Unexpected error in stream processing:', error); - - if (error.code === 'ERR_STREAM_PREMATURE_CLOSE') { - abortController.abort(); - return undefined; - } - - throw error; +export function setupSessionUsageHook(server: any, config: any): void { + // Session usage tracking is now handled in setupAgentProcessingHook } /** - * Read session usage from stream in background + * Setup error payload hook + * @deprecated This functionality is now integrated in setupAgentProcessingHook */ -function readSessionUsageFromStream(stream: ReadableStream, sessionId: string): void { - const read = async (stream: ReadableStream) => { - const reader = stream.getReader(); - try { - while (true) { - const { done, value } = await reader.read(); - if (done) break; - - const dataStr = new TextDecoder().decode(value); - if (!dataStr.startsWith("event: message_delta")) { - continue; - } - - const str = dataStr.slice(27); - try { - const message = JSON.parse(str); - sessionUsageCache.put(sessionId, message.usage); - } catch { - // Ignore parsing errors - } - } - } catch (readError: any) { - if (readError.name === 'AbortError' || readError.code === 'ERR_STREAM_PREMATURE_CLOSE') { - console.error('Background read stream closed prematurely'); - } else { - console.error('Error in background stream reading:', readError); - } - } finally { - reader.releaseLock(); - } - }; - - read(stream); +export function setupErrorPayloadHook(server: any): void { + // Error payload handling is now integrated in setupAgentProcessingHook } \ No newline at end of file diff --git a/src/utils/streamUtils.ts b/src/utils/streamUtils.ts new file mode 100644 index 00000000..a32c131d --- /dev/null +++ b/src/utils/streamUtils.ts @@ -0,0 +1,234 @@ +import { sessionUsageCache } from "./cache"; +import { StreamMetadata, PSSEResponse, StreamErrorInfo } from "./types"; + +/** + * Pads a number with leading zero if needed + */ +export function padZero(num: number): string { + return (num > 9 ? "" : "0") + num; +} + +/** + * Generates log file names with timestamp + */ +export function generateLogFileName(time?: Date, index?: number): string { + if (!time) { + time = new Date(); + } + + const yearAndMonth = time.getFullYear() + "" + padZero(time.getMonth() + 1); + const day = padZero(time.getDate()); + const hour = padZero(time.getHours()); + const minute = padZero(time.getMinutes()); + const second = padZero(time.getSeconds()); + + return `./logs/ccr-${yearAndMonth}${day}${hour}${minute}${second}${index ? `_${index}` : ''}.log`; +} + +/** + * Parse SSE content to extract structured information + */ +export function parseSSEContent(events: string[]): string | PSSEResponse { + try { + const messages = []; + for (const event of events) { + const lines = event.split('\n'); + for (const line of lines) { + if (line.startsWith('data: ') && line.length > 6) { + try { + const data = JSON.parse(line.substring(6)); + if (data.type === 'content_block_delta' && data.delta?.text) { + messages.push(data.delta.text); + } else if (data.type === 'content_block_start' && data.content_block?.type === 'text') { + messages.push(data.content_block.text); + } + } catch (e) { + // Ignore parsing errors and continue processing other data + } + } + } + } + + if (messages.length > 0) { + return { + type: "parsed_sse_response", + body: messages.join(''), + summary: { + totalEvents: events.length, + hasUsage: messages.some(m => m.type === 'usage'), + textEvents: messages.filter(m => m.type === 'text').length + } + }; + } + } catch (parseError) { + console.warn('Failed to parse SSE content:', parseError.message); + } + + return events; +} + +/** + * Create a wrapped stream that logs complete response content + */ +export function createLoggingWrappedStream( + originalStream: ReadableStream, + req: any, + reply: any +): ReadableStream { + const loggedChunks: string[] = []; + const startTime = Date.now(); + + return new ReadableStream({ + async start(controller) { + const reader = originalStream.getReader(); + + try { + while (true) { + const { done, value } = await reader.read(); + + if (done) { + await handleStreamCompletion(loggedChunks, startTime, req, reply, controller); + break; + } + + const chunkText = new TextDecoder().decode(value); + loggedChunks.push(chunkText); + controller.enqueue(value); + } + } catch (error) { + handleStreamError(error as Error, loggedChunks, startTime, req, controller); + } + } + }); +} + +/** + * Handle stream completion and logging + */ +async function handleStreamCompletion( + loggedChunks: string[], + startTime: number, + req: any, + reply: any, + controller: ReadableStreamDefaultController +): Promise { + const endTime = Date.now(); + const duration = endTime - startTime; + + req.log.info({ + streamCompleteResponse: { + type: "ReadableStream_complete_response", + content: parseSSEContent(loggedChunks), + metadata: createStreamMetadata(loggedChunks.length, duration, req, reply), + }, + msg: "Complete stream response" + }, "Stream response completed - full body"); + + controller.close(); +} + +/** + * Handle stream errors + */ +function handleStreamError( + error: Error, + loggedChunks: string[], + startTime: number, + req: any, + controller: ReadableStreamDefaultController +): void { + const endTime = Date.now(); + const duration = endTime - startTime; + + req.log.error({ + streamError: createStreamErrorInfo(error, loggedChunks.length, duration, req), + }, "Stream reading error"); + + controller.error(error); +} + +/** + * Create stream metadata for logging + */ +function createStreamMetadata( + totalChunks: number, + duration: number, + req: any, + reply: any +): StreamMetadata { + return { + totalChunks, + duration, + method: req.method, + url: req.url, + statusCode: reply.statusCode, + timestamp: new Date().toISOString() + }; +} + +/** + * Create stream error information for logging + */ +function createStreamErrorInfo( + error: Error, + chunksCollected: number, + duration: number, + req: any +): StreamErrorInfo { + return { + error: error.message, + stack: error.stack || '', + metadata: { + chunksCollected, + duration, + method: req.method, + url: req.url, + timestamp: new Date().toISOString() + } + }; +} + +/** + * Read session usage from stream in background + */ +export function readSessionUsageFromStream(stream: ReadableStream, sessionId: string): void { + const read = async (stream: ReadableStream) => { + const reader = stream.getReader(); + try { + while (true) { + const { done, value } = await reader.read(); + if (done) break; + + const dataStr = new TextDecoder().decode(value); + if (!dataStr.startsWith("event: message_delta")) { + continue; + } + + const str = dataStr.slice(27); + try { + const message = JSON.parse(str); + sessionUsageCache.put(sessionId, message.usage); + } catch { + // Ignore parsing errors + } + } + } catch (readError: any) { + handleBackgroundReadError(readError); + } finally { + reader.releaseLock(); + } + }; + + read(stream); +} + +/** + * Handle background read stream errors + */ +function handleBackgroundReadError(readError: any): void { + if (readError.name === 'AbortError' || readError.code === 'ERR_STREAM_PREMATURE_CLOSE') { + console.error('Background read stream closed prematurely'); + } else { + console.error('Error in background stream reading:', readError); + } +} \ No newline at end of file diff --git a/src/utils/types.ts b/src/utils/types.ts new file mode 100644 index 00000000..56b0e2a7 --- /dev/null +++ b/src/utils/types.ts @@ -0,0 +1,73 @@ +import { IAgent } from "../agents/type"; + +/** + * Agent processing state for handling tool calls + */ +export interface AgentProcessingState { + currentAgent?: IAgent; + currentToolIndex: number; + currentToolName: string; + currentToolArgs: string; + currentToolId: string; + toolMessages: ToolMessage[]; + assistantMessages: AssistantMessage[]; +} + +/** + * Tool message structure + */ +export interface ToolMessage { + tool_use_id: string; + type: "tool_result"; + content: string; +} + +/** + * Assistant message structure for tool use + */ +export interface AssistantMessage { + type: "tool_use"; + id: string; + name: string; + input: any; +} + +/** + * Stream processing metadata + */ +export interface StreamMetadata { + totalChunks: number; + duration: number; + method: string; + url: string; + statusCode: number; + timestamp: string; +} + +/** + * Parsed SSE response structure + */ +export interface PSSEResponse { + type: "parsed_sse_response"; + body: string; + summary: { + totalEvents: number; + hasUsage: boolean; + textEvents: number; + }; +} + +/** + * Stream error information + */ +export interface StreamErrorInfo { + error: string; + stack: string; + metadata: { + chunksCollected: number; + duration: number; + method: string; + url: string; + timestamp: string; + }; +} \ No newline at end of file From 1b27d61ff6f8fd7ebf65157c55c1724c25644ded Mon Sep 17 00:00:00 2001 From: Fu Gui Date: Wed, 26 Nov 2025 16:53:02 +0800 Subject: [PATCH 3/6] refactor: consolidate agent processing and stream utilities into hooks module MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Move agent processing logic from agentProcessor.ts into hooks.ts - Consolidate stream utilities from streamUtils.ts into hooks.ts - Remove duplicate type definitions by integrating into hooks.ts - Add new SSEParserUtils.ts for SSE parsing utilities - Update Claude settings to allow del command for file operations This refactoring improves code organization by centralizing related functionality and reducing module fragmentation while maintaining all existing capabilities. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- .claude/settings.local.json | 3 +- src/utils/SSEParserUtils.ts | 295 +++++++++++++++++ src/utils/agentProcessor.ts | 349 -------------------- src/utils/hooks.ts | 617 +++++++++++++++++++++++++++++++++++- src/utils/streamUtils.ts | 234 -------------- src/utils/types.ts | 73 ----- 6 files changed, 908 insertions(+), 663 deletions(-) create mode 100644 src/utils/SSEParserUtils.ts delete mode 100644 src/utils/agentProcessor.ts delete mode 100644 src/utils/streamUtils.ts delete mode 100644 src/utils/types.ts diff --git a/.claude/settings.local.json b/.claude/settings.local.json index 5f67ac46..4c7727f9 100644 --- a/.claude/settings.local.json +++ b/.claude/settings.local.json @@ -9,7 +9,8 @@ "Bash(Select-Object -Last 30)", "Bash(git log:*)", "Bash(git add:*)", - "Bash(git commit:*)" + "Bash(git commit:*)", + "Bash(del:*)" ], "deny": [], "ask": [] diff --git a/src/utils/SSEParserUtils.ts b/src/utils/SSEParserUtils.ts new file mode 100644 index 00000000..7131807e --- /dev/null +++ b/src/utils/SSEParserUtils.ts @@ -0,0 +1,295 @@ +interface TextContent { + type: 'text'; + text: string; +} + +interface ToolUseContent { + type: 'tool_use'; + id: string; + name: string; + input: any; + partialInput?: string; +} + +interface RedactedThinkingContent { + type: 'redacted_thinking'; + text: string; +} + +type ContentBlock = TextContent | ToolUseContent | RedactedThinkingContent; + +interface CompleteMessage { + id: string | null; + role: string | null; + content: ContentBlock[]; + model: string | null; +} + +interface SSEData { + type: string; + message?: { + id: string; + role: string; + model: string; + content?: any[]; + }; + index?: number; + content_block?: { + type: string; + text?: string; + id?: string; + name?: string; + input?: any; + }; + delta?: { + type?: string; + text?: string; + partial_json?: string; + }; +} + +class SSEMessageAssembler { + private currentMessage: CompleteMessage | null = null; + private currentContentBlock: ContentBlock | null = null; + private currentIndex: number | null = null; + + public reset(): void { + this.currentMessage = null; + this.currentContentBlock = null; + this.currentIndex = null; + } + + public processEvent(eventData: string): CompleteMessage | null { + try { + const data: SSEData = JSON.parse(eventData); + + switch (data.type) { + case 'message_start': + return this.handleMessageStart(data); + case 'content_block_start': + this.handleContentBlockStart(data); + break; + case 'content_block_delta': + this.handleContentBlockDelta(data); + break; + case 'content_block_stop': + this.handleContentBlockStop(data); + break; + case 'message_stop': + return this.handleMessageStop(data); + case 'ping': + // 忽略心跳包，但保持当前状态 + break; + default: + console.warn('未知的 SSE 事件类型:', data.type); + } + } catch (error) { + console.error('解析 SSE 事件数据失败:', error, '原始数据:', eventData); + } + + return null; + } + + private handleMessageStart(data: SSEData): null { + this.currentMessage = { + id: data.message?.id || null, + role: data.message?.role || null, + content: [], + model: data.message?.model || null + }; + return null; + } + + private handleContentBlockStart(data: SSEData): void { + if (!this.currentMessage) return; + + this.currentIndex = data.index ?? 0; + + const contentBlock = data.content_block; + if (!contentBlock) return; + + switch (contentBlock.type) { + case 'text': + this.currentContentBlock = { + type: 'text', + text: contentBlock.text || '' + }; + break; + + case 'tool_use': + this.currentContentBlock = { + type: 'tool_use', + id: contentBlock.id || '', + name: contentBlock.name || '', + input: contentBlock.input || {}, + partialInput: '' + }; + break; + + case 'redacted_thinking': + this.currentContentBlock = { + type: 'redacted_thinking', + text: contentBlock.text || '' + }; + break; + + default: + console.warn('未知的内容块类型:', contentBlock.type); + this.currentContentBlock = null; + return; + } + + if (this.currentContentBlock && this.currentIndex !== null) { + this.currentMessage.content[this.currentIndex] = this.currentContentBlock; + } + } + + private handleContentBlockDelta(data: SSEData): void { + if (!this.currentContentBlock || !this.currentMessage || this.currentIndex === null) return; + + const delta = data.delta; + if (!delta) return; + + switch (this.currentContentBlock.type) { + case 'text': + if (delta.text) { + this.currentContentBlock.text += delta.text; + } + break; + + case 'tool_use': + if (delta.type === 'input_json_delta' && delta.partial_json) { + // 累积部分 JSON 输入 + if (!this.currentContentBlock.partialInput) { + this.currentContentBlock.partialInput = ''; + } + this.currentContentBlock.partialInput += delta.partial_json; + + // 尝试解析为完整对象（可能失败，因为 JSON 还不完整） + try { + this.currentContentBlock.input = JSON.parse(this.currentContentBlock.partialInput); + } catch (e) { + // JSON 还不完整，继续累积 + } + } + break; + + case 'redacted_thinking': + if (delta.text) { + this.currentContentBlock.text += delta.text; + } + break; + } + + // 更新消息中的对应块 + this.currentMessage.content[this.currentIndex] = this.currentContentBlock; + } + + private handleContentBlockStop(data: SSEData): void { + if (this.currentContentBlock?.type === 'tool_use' && this.currentContentBlock.partialInput) { + // 对于工具调用，在块结束时尝试最终解析 + try { + this.currentContentBlock.input = JSON.parse(this.currentContentBlock.partialInput); + delete this.currentContentBlock.partialInput; + } catch (e) { + console.error('工具调用输入 JSON 解析失败:', e, '原始数据:', this.currentContentBlock.partialInput); + // 保留部分输入以便调试 + } + } + + this.currentContentBlock = null; + this.currentIndex = null; + } + + private handleMessageStop(data: SSEData): CompleteMessage | null { + if (!this.currentMessage) return null; + + const completeMessage = { ...this.currentMessage }; + this.reset(); + + return completeMessage; + } + + public getCurrentState(): { message: CompleteMessage | null; currentBlock: ContentBlock | null } { + return { + message: this.currentMessage ? { ...this.currentMessage } : null, + currentBlock: this.currentContentBlock ? { ...this.currentContentBlock } : null + }; + } +} + +// 主函数 - 直接使用这个函数 +export function parseSSEContent(events: string[]): CompleteMessage | null { + const assembler = new SSEMessageAssembler(); + let finalMessage: CompleteMessage | null = null; + + for (const event of events) { + // 清理事件数据（移除 "data: " 前缀等） + const cleanEvent = event.replace(/^data: /, '').trim(); + + // 跳过空行和注释 + if (!cleanEvent || cleanEvent.startsWith(':')) continue; + + const result = assembler.processEvent(cleanEvent); + if (result) { + finalMessage = result; + } + } + + // 如果没有收到 message_stop 事件，但想强制获取当前状态 + if (!finalMessage) { + const state = assembler.getCurrentState(); + finalMessage = state.message; + } + + return finalMessage; +} + +// 工具函数：从完整消息中提取特定类型的内容 +export function extractContentByType(message: CompleteMessage, type: 'text' | 'tool_use' | 'redacted_thinking'): any[] { + return message.content.filter(block => block.type === type); +} + +// 工具函数：获取所有文本内容（合并所有 text 块） +export function getAllTextContent(message: CompleteMessage): string { + return message.content + .filter((block): block is TextContent => block.type === 'text') + .map(block => block.text) + .join('\n'); +} + +// 工具函数：获取所有工具调用 +export function getAllToolCalls(message: CompleteMessage): ToolUseContent[] { + return message.content.filter((block): block is ToolUseContent => block.type === 'tool_use'); +} + +export function getFullContent(events: string[]): string { + const result = parseSSEContent(events); + if( result != null ) + return getAllTextContent(result) + '\n' + JSON.stringify( getAllToolCalls(result) ); + return JSON.stringify(events); +} + + +// 使用示例 +/* +const sseEvents = [ + `data: {"type": "message_start", "message": {"id": "msg_123", "role": "assistant", "model": "claude-3-sonnet"}}`, + `data: {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}`, + `data: {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "Hello"}}`, + `data: {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": " World"}}`, + `data: {"type": "content_block_stop", "index": 0}`, + `data: {"type": "content_block_start", "index": 1, "content_block": {"type": "tool_use", "id": "toolu_001", "name": "get_weather"}}`, + `data: {"type": "content_block_delta", "index": 1, "delta": {"type": "input_json_delta", "partial_json": "{\\"location\\":\\"Beijing\\""}}`, + `data: {"type": "content_block_delta", "index": 1, "delta": {"type": "input_json_delta", "partial_json": ",\\"unit\\":\\"celsius\\"}"}}`, + `data: {"type": "content_block_stop", "index": 1}`, + `data: {"type": "message_stop"}`, +]; + +const result = parseSSEContent(sseEvents); +if (result) { + console.log('完整消息:', JSON.stringify(result, null, 2)); + console.log('所有文本:', getAllTextContent(result)); + console.log('工具调用:', getAllToolCalls(result)); +} +*/ \ No newline at end of file diff --git a/src/utils/agentProcessor.ts b/src/utils/agentProcessor.ts deleted file mode 100644 index b385e1f8..00000000 --- a/src/utils/agentProcessor.ts +++ /dev/null @@ -1,349 +0,0 @@ -import agentsManager from "../agents"; -import { SSEParserTransform } from "./SSEParser.transform"; -import { SSESerializerTransform } from "./SSESerializer.transform"; -import { rewriteStream } from "./rewriteStream"; -import JSON5 from "json5"; -import { - AgentProcessingState, - AssistantMessage, - ToolMessage -} from "./types"; - -/** - * Handle agent stream processing with tool calls - */ -export function handleAgentStreamProcessing( - payload: ReadableStream, - req: any, - config: any -): ReadableStream { - const abortController = new AbortController(); - const eventStream = payload.pipeThrough(new SSEParserTransform()); - const agentState = createInitialAgentState(); - - return rewriteStream(eventStream, async (data, controller) => { - try { - const result = await processStreamEvent(data, agentState, req, config); - - if (result === 'continue') { - return undefined; - } - - if (result === 'retry_with_tools') { - await retryRequestWithToolResults(agentState, req, config, controller, abortController); - return undefined; - } - - return data; - } catch (error: any) { - handleStreamProcessingError(error, abortController); - } - }).pipeThrough(new SSESerializerTransform()); -} - -/** - * Create initial agent processing state - */ -function createInitialAgentState(): AgentProcessingState { - return { - currentAgent: undefined, - currentToolIndex: -1, - currentToolName: '', - currentToolArgs: '', - currentToolId: '', - toolMessages: [], - assistantMessages: [] - }; -} - -/** - * Process individual stream events - */ -async function processStreamEvent( - data: any, - agentState: AgentProcessingState, - req: any, - config: any -): Promise<'continue' | 'retry_with_tools' | 'pass_through'> { - // Handle tool call start - if (isToolCallStart(data)) { - const agent = findAgentForTool(data.data.content_block.name, req); - if (agent) { - updateAgentStateForToolStart(agentState, agent, data); - return 'continue'; - } - } - - // Collect tool arguments - if (isToolArgumentCollection(data, agentState)) { - collectToolArguments(agentState, data); - return 'continue'; - } - - // Handle tool call completion - if (isToolCallCompletion(data, agentState)) { - await processToolCall(agentState, req, config); - return 'continue'; - } - - // Handle message delta with tool results - if (isMessageDeltaWithToolResults(data, agentState)) { - return 'retry_with_tools'; - } - - return 'pass_through'; -} - -/** - * Check if event is tool call start - */ -function isToolCallStart(data: any): boolean { - return data.event === 'content_block_start' && - data?.data?.content_block?.name; -} - -/** - * Find agent that can handle the tool - */ -function findAgentForTool(toolName: string, req: any): string | undefined { - return req.agents.find((name: string) => - agentsManager.getAgent(name)?.tools.get(toolName) - ); -} - -/** - * Update agent state when tool call starts - */ -function updateAgentStateForToolStart( - agentState: AgentProcessingState, - agentName: string, - data: any -): void { - const agent = agentsManager.getAgent(agentName); - agentState.currentAgent = agent; - agentState.currentToolIndex = data.data.index; - agentState.currentToolName = data.data.content_block.name; - agentState.currentToolId = data.data.content_block.id; -} - -/** - * Check if event is for collecting tool arguments - */ -function isToolArgumentCollection(data: any, agentState: AgentProcessingState): boolean { - return agentState.currentToolIndex > -1 && - data.data.index === agentState.currentToolIndex && - data.data?.delta?.type === 'input_json_delta'; -} - -/** - * Collect tool arguments from delta - */ -function collectToolArguments(agentState: AgentProcessingState, data: any): void { - agentState.currentToolArgs += data.data?.delta?.partial_json; -} - -/** - * Check if event is tool call completion - */ -function isToolCallCompletion(data: any, agentState: AgentProcessingState): boolean { - return agentState.currentToolIndex > -1 && - data.data.index === agentState.currentToolIndex && - data.data.type === 'content_block_stop'; -} - -/** - * Process a completed tool call - */ -async function processToolCall( - agentState: AgentProcessingState, - req: any, - config: any -): Promise { - try { - const args = JSON5.parse(agentState.currentToolArgs); - - const assistantMessage = createAssistantMessage(agentState); - agentState.assistantMessages.push(assistantMessage); - - const toolResult = await executeTool(agentState, args, req, config); - const toolMessage = createToolMessage(agentState.currentToolId, toolResult); - agentState.toolMessages.push(toolMessage); - - resetAgentToolState(agentState); - } catch (error) { - console.error('Error processing tool call:', error); - } -} - -/** - * Create assistant message for tool use - */ -function createAssistantMessage(agentState: AgentProcessingState): AssistantMessage { - return { - type: "tool_use", - id: agentState.currentToolId, - name: agentState.currentToolName, - input: JSON5.parse(agentState.currentToolArgs) - }; -} - -/** - * Execute tool and get result - */ -async function executeTool( - agentState: AgentProcessingState, - args: any, - req: any, - config: any -): Promise { - const tool = agentState.currentAgent?.tools.get(agentState.currentToolName); - if (!tool) { - throw new Error(`Tool ${agentState.currentToolName} not found`); - } - - return await tool.handler(args, { req, config }); -} - -/** - * Create tool result message - */ -function createToolMessage(toolUseId: string, toolResult: string): ToolMessage { - return { - tool_use_id: toolUseId, - type: "tool_result", - content: toolResult - }; -} - -/** - * Reset agent tool-specific state - */ -function resetAgentToolState(agentState: AgentProcessingState): void { - agentState.currentAgent = undefined; - agentState.currentToolIndex = -1; - agentState.currentToolName = ''; - agentState.currentToolArgs = ''; - agentState.currentToolId = ''; -} - -/** - * Check if event is message delta with tool results - */ -function isMessageDeltaWithToolResults(data: any, agentState: AgentProcessingState): boolean { - return data.event === 'message_delta' && agentState.toolMessages.length > 0; -} - -/** - * Retry request with tool results - */ -async function retryRequestWithToolResults( - agentState: AgentProcessingState, - req: any, - config: any, - controller: any, - abortController: AbortController -): Promise { - addToolMessagesToRequest(req, agentState); - - const response = await makeRetryRequest(req, config); - if (!response.ok) { - return; - } - - await processRetryResponse(response, controller, abortController); -} - -/** - * Add tool messages to request body - */ -function addToolMessagesToRequest(req: any, agentState: AgentProcessingState): void { - req.body.messages.push({ - role: 'assistant', - content: agentState.assistantMessages - }); - - req.body.messages.push({ - role: 'user', - content: agentState.toolMessages - }); -} - -/** - * Make retry request with tool results - */ -async function makeRetryRequest(req: any, config: any): Promise { - return await fetch(`http://127.0.0.1:${config.PORT || 3456}/v1/messages`, { - method: "POST", - headers: { - 'x-api-key': config.APIKEY, - 'content-type': 'application/json', - }, - body: JSON.stringify(req.body), - }); -} - -/** - * Process retry response stream - */ -async function processRetryResponse( - response: Response, - controller: any, - abortController: AbortController -): Promise { - const stream = response.body!.pipeThrough(new SSEParserTransform()); - const reader = stream.getReader(); - - while (true) { - try { - const { value, done } = await reader.read(); - if (done) { - break; - } - - if (shouldSkipEvent(value)) { - continue; - } - - if (!controller.desiredSize) { - break; - } - - controller.enqueue(value); - } catch (readError: any) { - handleRetryStreamError(readError, abortController); - break; - } - } -} - -/** - * Check if event should be skipped - */ -function shouldSkipEvent(value: any): boolean { - return ['message_start', 'message_stop'].includes(value.event); -} - -/** - * Handle retry stream errors - */ -function handleRetryStreamError(readError: any, abortController: AbortController): void { - if (readError.name === 'AbortError' || readError.code === 'ERR_STREAM_PREMATURE_CLOSE') { - abortController.abort(); - } else { - throw readError; - } -} - -/** - * Handle stream processing errors - */ -function handleStreamProcessingError(error: any, abortController: AbortController): void { - console.error('Unexpected error in stream processing:', error); - - if (error.code === 'ERR_STREAM_PREMATURE_CLOSE') { - abortController.abort(); - return; - } - - throw error; -} \ No newline at end of file diff --git a/src/utils/hooks.ts b/src/utils/hooks.ts index 219e9be0..27254eb9 100644 --- a/src/utils/hooks.ts +++ b/src/utils/hooks.ts @@ -4,12 +4,83 @@ import agentsManager from "../agents"; import { sessionUsageCache } from "./cache"; import { apiKeyAuth } from "../middleware/auth"; import { router } from "./router"; -import { - createLoggingWrappedStream, - readSessionUsageFromStream, - parseSSEContent -} from "./streamUtils"; -import { handleAgentStreamProcessing } from "./agentProcessor"; +import { SSEParserTransform } from "./SSEParser.transform"; +import { SSESerializerTransform } from "./SSESerializer.transform"; +import { rewriteStream } from "./rewriteStream"; +import {getFullContent} from "./SSEParserUtils" +import JSON5 from "json5"; + +/** + * Agent processing state for handling tool calls + */ +interface AgentProcessingState { + currentAgent?: IAgent; + currentToolIndex: number; + currentToolName: string; + currentToolArgs: string; + currentToolId: string; + toolMessages: ToolMessage[]; + assistantMessages: AssistantMessage[]; +} + +/** + * Tool message structure + */ +interface ToolMessage { + tool_use_id: string; + type: "tool_result"; + content: string; +} + +/** + * Assistant message structure for tool use + */ +interface AssistantMessage { + type: "tool_use"; + id: string; + name: string; + input: any; +} + +/** + * Stream processing metadata + */ +interface StreamMetadata { + totalChunks: number; + duration: number; + method: string; + url: string; + statusCode: number; + timestamp: string; +} + +/** + * Parsed SSE response structure + */ +interface PSSEResponse { + type: "parsed_sse_response"; + body: string; + summary: { + totalEvents: number; + hasUsage: boolean; + textEvents: number; + }; +} + +/** + * Stream error information + */ +interface StreamErrorInfo { + error: string; + stack: string; + metadata: { + chunksCollected: number; + duration: number; + method: string; + url: string; + timestamp: string; + }; +} /** * Setup request logging hook @@ -265,4 +336,538 @@ export function setupSessionUsageHook(server: any, config: any): void { */ export function setupErrorPayloadHook(server: any): void { // Error payload handling is now integrated in setupAgentProcessingHook +} + +// ============= Stream Utilities ============= + +/** + * Pads a number with leading zero if needed + */ +function padZero(num: number): string { + return (num > 9 ? "" : "0") + num; +} + +/** + * Generates log file names with timestamp + */ +export function generateLogFileName(time?: Date, index?: number): string { + if (!time) { + time = new Date(); + } + + const yearAndMonth = time.getFullYear() + "" + padZero(time.getMonth() + 1); + const day = padZero(time.getDate()); + const hour = padZero(time.getHours()); + const minute = padZero(time.getMinutes()); + const second = padZero(time.getSeconds()); + + return `./logs/ccr-${yearAndMonth}${day}${hour}${minute}${second}${index ? `_${index}` : ''}.log`; +} + + +/** + * Create a wrapped stream that logs complete response content + */ +function createLoggingWrappedStream( + originalStream: ReadableStream, + req: any, + reply: any +): ReadableStream { + const loggedChunks: string[] = []; + const startTime = Date.now(); + + return new ReadableStream({ + async start(controller) { + const reader = originalStream.getReader(); + + try { + while (true) { + const { done, value } = await reader.read(); + + if (done) { + await handleStreamCompletion(loggedChunks, startTime, req, reply, controller); + break; + } + + const chunkText = new TextDecoder().decode(value); + loggedChunks.push(chunkText); + controller.enqueue(value); + } + } catch (error) { + handleStreamError(error as Error, loggedChunks, startTime, req, controller); + } + } + }); +} + +/** + * Handle stream completion and logging + */ +async function handleStreamCompletion( + loggedChunks: string[], + startTime: number, + req: any, + reply: any, + controller: ReadableStreamDefaultController +): Promise { + const endTime = Date.now(); + const duration = endTime - startTime; + + req.log.info({ + streamCompleteResponse: { + type: "ReadableStream_complete_response", + content: getFullContent(loggedChunks), + metadata: createStreamMetadata(loggedChunks.length, duration, req, reply), + }, + msg: "Complete stream response" + }, "Stream response completed - full body"); + + controller.close(); +} + +/** + * Handle stream errors + */ +function handleStreamError( + error: Error, + loggedChunks: string[], + startTime: number, + req: any, + controller: ReadableStreamDefaultController +): void { + const endTime = Date.now(); + const duration = endTime - startTime; + + req.log.error({ + streamError: createStreamErrorInfo(error, loggedChunks.length, duration, req), + }, "Stream reading error"); + + controller.error(error); +} + +/** + * Create stream metadata for logging + */ +function createStreamMetadata( + totalChunks: number, + duration: number, + req: any, + reply: any +): StreamMetadata { + return { + totalChunks, + duration, + method: req.method, + url: req.url, + statusCode: reply.statusCode, + timestamp: new Date().toISOString() + }; +} + +/** + * Create stream error information for logging + */ +function createStreamErrorInfo( + error: Error, + chunksCollected: number, + duration: number, + req: any +): StreamErrorInfo { + return { + error: error.message, + stack: error.stack || '', + metadata: { + chunksCollected, + duration, + method: req.method, + url: req.url, + timestamp: new Date().toISOString() + } + }; +} + +/** + * Read session usage from stream in background + */ +function readSessionUsageFromStream(stream: ReadableStream, sessionId: string): void { + const read = async (stream: ReadableStream) => { + const reader = stream.getReader(); + try { + while (true) { + const { done, value } = await reader.read(); + if (done) break; + + const dataStr = new TextDecoder().decode(value); + if (!dataStr.startsWith("event: message_delta")) { + continue; + } + + const str = dataStr.slice(27); + try { + const message = JSON.parse(str); + sessionUsageCache.put(sessionId, message.usage); + } catch { + // Ignore parsing errors + } + } + } catch (readError: any) { + handleBackgroundReadError(readError); + } finally { + reader.releaseLock(); + } + }; + + read(stream); +} + +/** + * Handle background read stream errors + */ +function handleBackgroundReadError(readError: any): void { + if (readError.name === 'AbortError' || readError.code === 'ERR_STREAM_PREMATURE_CLOSE') { + console.error('Background read stream closed prematurely'); + } else { + console.error('Error in background stream reading:', readError); + } +} + +// ============= Agent Stream Processing ============= + +/** + * Handle agent stream processing with tool calls + */ +function handleAgentStreamProcessing( + payload: ReadableStream, + req: any, + config: any +): ReadableStream { + const abortController = new AbortController(); + const eventStream = payload.pipeThrough(new SSEParserTransform()); + const agentState = createInitialAgentState(); + + return rewriteStream(eventStream, async (data, controller) => { + try { + const result = await processStreamEvent(data, agentState, req, config); + + if (result === 'continue') { + return undefined; + } + + if (result === 'retry_with_tools') { + await retryRequestWithToolResults(agentState, req, config, controller, abortController); + return undefined; + } + + return data; + } catch (error: any) { + handleStreamProcessingError(error, abortController); + } + }).pipeThrough(new SSESerializerTransform()); +} + +/** + * Create initial agent processing state + */ +function createInitialAgentState(): AgentProcessingState { + return { + currentAgent: undefined, + currentToolIndex: -1, + currentToolName: '', + currentToolArgs: '', + currentToolId: '', + toolMessages: [], + assistantMessages: [] + }; +} + +/** + * Process individual stream events + */ +async function processStreamEvent( + data: any, + agentState: AgentProcessingState, + req: any, + config: any +): Promise<'continue' | 'retry_with_tools' | 'pass_through'> { + // Handle tool call start + if (isToolCallStart(data)) { + const agent = findAgentForTool(data.data.content_block.name, req); + if (agent) { + updateAgentStateForToolStart(agentState, agent, data); + return 'continue'; + } + } + + // Collect tool arguments + if (isToolArgumentCollection(data, agentState)) { + collectToolArguments(agentState, data); + return 'continue'; + } + + // Handle tool call completion + if (isToolCallCompletion(data, agentState)) { + await processToolCall(agentState, req, config); + return 'continue'; + } + + // Handle message delta with tool results + if (isMessageDeltaWithToolResults(data, agentState)) { + return 'retry_with_tools'; + } + + return 'pass_through'; +} + +/** + * Check if event is tool call start + */ +function isToolCallStart(data: any): boolean { + return data.event === 'content_block_start' && + data?.data?.content_block?.name; +} + +/** + * Find agent that can handle the tool + */ +function findAgentForTool(toolName: string, req: any): string | undefined { + return req.agents.find((name: string) => + agentsManager.getAgent(name)?.tools.get(toolName) + ); +} + +/** + * Update agent state when tool call starts + */ +function updateAgentStateForToolStart( + agentState: AgentProcessingState, + agentName: string, + data: any +): void { + const agent = agentsManager.getAgent(agentName); + agentState.currentAgent = agent; + agentState.currentToolIndex = data.data.index; + agentState.currentToolName = data.data.content_block.name; + agentState.currentToolId = data.data.content_block.id; +} + +/** + * Check if event is for collecting tool arguments + */ +function isToolArgumentCollection(data: any, agentState: AgentProcessingState): boolean { + return agentState.currentToolIndex > -1 && + data.data.index === agentState.currentToolIndex && + data.data?.delta?.type === 'input_json_delta'; +} + +/** + * Collect tool arguments from delta + */ +function collectToolArguments(agentState: AgentProcessingState, data: any): void { + agentState.currentToolArgs += data.data?.delta?.partial_json; +} + +/** + * Check if event is tool call completion + */ +function isToolCallCompletion(data: any, agentState: AgentProcessingState): boolean { + return agentState.currentToolIndex > -1 && + data.data.index === agentState.currentToolIndex && + data.data.type === 'content_block_stop'; +} + +/** + * Process a completed tool call + */ +async function processToolCall( + agentState: AgentProcessingState, + req: any, + config: any +): Promise { + try { + const args = JSON5.parse(agentState.currentToolArgs); + + const assistantMessage = createAssistantMessage(agentState); + agentState.assistantMessages.push(assistantMessage); + + const toolResult = await executeTool(agentState, args, req, config); + const toolMessage = createToolMessage(agentState.currentToolId, toolResult); + agentState.toolMessages.push(toolMessage); + + resetAgentToolState(agentState); + } catch (error) { + console.error('Error processing tool call:', error); + } +} + +/** + * Create assistant message for tool use + */ +function createAssistantMessage(agentState: AgentProcessingState): AssistantMessage { + return { + type: "tool_use", + id: agentState.currentToolId, + name: agentState.currentToolName, + input: JSON5.parse(agentState.currentToolArgs) + }; +} + +/** + * Execute tool and get result + */ +async function executeTool( + agentState: AgentProcessingState, + args: any, + req: any, + config: any +): Promise { + const tool = agentState.currentAgent?.tools.get(agentState.currentToolName); + if (!tool) { + throw new Error(`Tool ${agentState.currentToolName} not found`); + } + + return await tool.handler(args, { req, config }); +} + +/** + * Create tool result message + */ +function createToolMessage(toolUseId: string, toolResult: string): ToolMessage { + return { + tool_use_id: toolUseId, + type: "tool_result", + content: toolResult + }; +} + +/** + * Reset agent tool-specific state + */ +function resetAgentToolState(agentState: AgentProcessingState): void { + agentState.currentAgent = undefined; + agentState.currentToolIndex = -1; + agentState.currentToolName = ''; + agentState.currentToolArgs = ''; + agentState.currentToolId = ''; +} + +/** + * Check if event is message delta with tool results + */ +function isMessageDeltaWithToolResults(data: any, agentState: AgentProcessingState): boolean { + return data.event === 'message_delta' && agentState.toolMessages.length > 0; +} + +/** + * Retry request with tool results + */ +async function retryRequestWithToolResults( + agentState: AgentProcessingState, + req: any, + config: any, + controller: any, + abortController: AbortController +): Promise { + addToolMessagesToRequest(req, agentState); + + const response = await makeRetryRequest(req, config); + if (!response.ok) { + return; + } + + await processRetryResponse(response, controller, abortController); +} + +/** + * Add tool messages to request body + */ +function addToolMessagesToRequest(req: any, agentState: AgentProcessingState): void { + req.body.messages.push({ + role: 'assistant', + content: agentState.assistantMessages + }); + + req.body.messages.push({ + role: 'user', + content: agentState.toolMessages + }); +} + +/** + * Make retry request with tool results + */ +async function makeRetryRequest(req: any, config: any): Promise { + return await fetch(`http://127.0.0.1:${config.PORT || 3456}/v1/messages`, { + method: "POST", + headers: { + 'x-api-key': config.APIKEY, + 'content-type': 'application/json', + }, + body: JSON.stringify(req.body), + }); +} + +/** + * Process retry response stream + */ +async function processRetryResponse( + response: Response, + controller: any, + abortController: AbortController +): Promise { + const stream = response.body!.pipeThrough(new SSEParserTransform()); + const reader = stream.getReader(); + + while (true) { + try { + const { value, done } = await reader.read(); + if (done) { + break; + } + + if (shouldSkipEvent(value)) { + continue; + } + + if (!controller.desiredSize) { + break; + } + + controller.enqueue(value); + } catch (readError: any) { + handleRetryStreamError(readError, abortController); + break; + } + } +} + +/** + * Check if event should be skipped + */ +function shouldSkipEvent(value: any): boolean { + return ['message_start', 'message_stop'].includes(value.event); +} + +/** + * Handle retry stream errors + */ +function handleRetryStreamError(readError: any, abortController: AbortController): void { + if (readError.name === 'AbortError' || readError.code === 'ERR_STREAM_PREMATURE_CLOSE') { + abortController.abort(); + } else { + throw readError; + } +} + +/** + * Handle stream processing errors + */ +function handleStreamProcessingError(error: any, abortController: AbortController): void { + console.error('Unexpected error in stream processing:', error); + + if (error.code === 'ERR_STREAM_PREMATURE_CLOSE') { + abortController.abort(); + return; + } + + throw error; } \ No newline at end of file diff --git a/src/utils/streamUtils.ts b/src/utils/streamUtils.ts deleted file mode 100644 index a32c131d..00000000 --- a/src/utils/streamUtils.ts +++ /dev/null @@ -1,234 +0,0 @@ -import { sessionUsageCache } from "./cache"; -import { StreamMetadata, PSSEResponse, StreamErrorInfo } from "./types"; - -/** - * Pads a number with leading zero if needed - */ -export function padZero(num: number): string { - return (num > 9 ? "" : "0") + num; -} - -/** - * Generates log file names with timestamp - */ -export function generateLogFileName(time?: Date, index?: number): string { - if (!time) { - time = new Date(); - } - - const yearAndMonth = time.getFullYear() + "" + padZero(time.getMonth() + 1); - const day = padZero(time.getDate()); - const hour = padZero(time.getHours()); - const minute = padZero(time.getMinutes()); - const second = padZero(time.getSeconds()); - - return `./logs/ccr-${yearAndMonth}${day}${hour}${minute}${second}${index ? `_${index}` : ''}.log`; -} - -/** - * Parse SSE content to extract structured information - */ -export function parseSSEContent(events: string[]): string | PSSEResponse { - try { - const messages = []; - for (const event of events) { - const lines = event.split('\n'); - for (const line of lines) { - if (line.startsWith('data: ') && line.length > 6) { - try { - const data = JSON.parse(line.substring(6)); - if (data.type === 'content_block_delta' && data.delta?.text) { - messages.push(data.delta.text); - } else if (data.type === 'content_block_start' && data.content_block?.type === 'text') { - messages.push(data.content_block.text); - } - } catch (e) { - // Ignore parsing errors and continue processing other data - } - } - } - } - - if (messages.length > 0) { - return { - type: "parsed_sse_response", - body: messages.join(''), - summary: { - totalEvents: events.length, - hasUsage: messages.some(m => m.type === 'usage'), - textEvents: messages.filter(m => m.type === 'text').length - } - }; - } - } catch (parseError) { - console.warn('Failed to parse SSE content:', parseError.message); - } - - return events; -} - -/** - * Create a wrapped stream that logs complete response content - */ -export function createLoggingWrappedStream( - originalStream: ReadableStream, - req: any, - reply: any -): ReadableStream { - const loggedChunks: string[] = []; - const startTime = Date.now(); - - return new ReadableStream({ - async start(controller) { - const reader = originalStream.getReader(); - - try { - while (true) { - const { done, value } = await reader.read(); - - if (done) { - await handleStreamCompletion(loggedChunks, startTime, req, reply, controller); - break; - } - - const chunkText = new TextDecoder().decode(value); - loggedChunks.push(chunkText); - controller.enqueue(value); - } - } catch (error) { - handleStreamError(error as Error, loggedChunks, startTime, req, controller); - } - } - }); -} - -/** - * Handle stream completion and logging - */ -async function handleStreamCompletion( - loggedChunks: string[], - startTime: number, - req: any, - reply: any, - controller: ReadableStreamDefaultController -): Promise { - const endTime = Date.now(); - const duration = endTime - startTime; - - req.log.info({ - streamCompleteResponse: { - type: "ReadableStream_complete_response", - content: parseSSEContent(loggedChunks), - metadata: createStreamMetadata(loggedChunks.length, duration, req, reply), - }, - msg: "Complete stream response" - }, "Stream response completed - full body"); - - controller.close(); -} - -/** - * Handle stream errors - */ -function handleStreamError( - error: Error, - loggedChunks: string[], - startTime: number, - req: any, - controller: ReadableStreamDefaultController -): void { - const endTime = Date.now(); - const duration = endTime - startTime; - - req.log.error({ - streamError: createStreamErrorInfo(error, loggedChunks.length, duration, req), - }, "Stream reading error"); - - controller.error(error); -} - -/** - * Create stream metadata for logging - */ -function createStreamMetadata( - totalChunks: number, - duration: number, - req: any, - reply: any -): StreamMetadata { - return { - totalChunks, - duration, - method: req.method, - url: req.url, - statusCode: reply.statusCode, - timestamp: new Date().toISOString() - }; -} - -/** - * Create stream error information for logging - */ -function createStreamErrorInfo( - error: Error, - chunksCollected: number, - duration: number, - req: any -): StreamErrorInfo { - return { - error: error.message, - stack: error.stack || '', - metadata: { - chunksCollected, - duration, - method: req.method, - url: req.url, - timestamp: new Date().toISOString() - } - }; -} - -/** - * Read session usage from stream in background - */ -export function readSessionUsageFromStream(stream: ReadableStream, sessionId: string): void { - const read = async (stream: ReadableStream) => { - const reader = stream.getReader(); - try { - while (true) { - const { done, value } = await reader.read(); - if (done) break; - - const dataStr = new TextDecoder().decode(value); - if (!dataStr.startsWith("event: message_delta")) { - continue; - } - - const str = dataStr.slice(27); - try { - const message = JSON.parse(str); - sessionUsageCache.put(sessionId, message.usage); - } catch { - // Ignore parsing errors - } - } - } catch (readError: any) { - handleBackgroundReadError(readError); - } finally { - reader.releaseLock(); - } - }; - - read(stream); -} - -/** - * Handle background read stream errors - */ -function handleBackgroundReadError(readError: any): void { - if (readError.name === 'AbortError' || readError.code === 'ERR_STREAM_PREMATURE_CLOSE') { - console.error('Background read stream closed prematurely'); - } else { - console.error('Error in background stream reading:', readError); - } -} \ No newline at end of file diff --git a/src/utils/types.ts b/src/utils/types.ts deleted file mode 100644 index 56b0e2a7..00000000 --- a/src/utils/types.ts +++ /dev/null @@ -1,73 +0,0 @@ -import { IAgent } from "../agents/type"; - -/** - * Agent processing state for handling tool calls - */ -export interface AgentProcessingState { - currentAgent?: IAgent; - currentToolIndex: number; - currentToolName: string; - currentToolArgs: string; - currentToolId: string; - toolMessages: ToolMessage[]; - assistantMessages: AssistantMessage[]; -} - -/** - * Tool message structure - */ -export interface ToolMessage { - tool_use_id: string; - type: "tool_result"; - content: string; -} - -/** - * Assistant message structure for tool use - */ -export interface AssistantMessage { - type: "tool_use"; - id: string; - name: string; - input: any; -} - -/** - * Stream processing metadata - */ -export interface StreamMetadata { - totalChunks: number; - duration: number; - method: string; - url: string; - statusCode: number; - timestamp: string; -} - -/** - * Parsed SSE response structure - */ -export interface PSSEResponse { - type: "parsed_sse_response"; - body: string; - summary: { - totalEvents: number; - hasUsage: boolean; - textEvents: number; - }; -} - -/** - * Stream error information - */ -export interface StreamErrorInfo { - error: string; - stack: string; - metadata: { - chunksCollected: number; - duration: number; - method: string; - url: string; - timestamp: string; - }; -} \ No newline at end of file From 5e755ba99bb7ae911eee965089ae1365a19a5de7 Mon Sep 17 00:00:00 2001 From: Fu Gui Date: Wed, 26 Nov 2025 17:27:05 +0800 Subject: [PATCH 4/6] fix: improve SSE parser with better error handling and add test coverage MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Fix text content extraction to handle empty content properly - Format tool call output with readable JSON indentation - Add proper error messages for failed SSE parsing - Add test file for SSE parser utilities - Update local settings with required command permissions 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- .claude/settings.local.json | 5 +- src/utils/SSEParserUtils.test.ts | 240 +++++++++++++++++++++++++++++++ src/utils/SSEParserUtils.ts | 30 +++- 3 files changed, 271 insertions(+), 4 deletions(-) create mode 100644 src/utils/SSEParserUtils.test.ts diff --git a/.claude/settings.local.json b/.claude/settings.local.json index 4c7727f9..7f17b2ff 100644 --- a/.claude/settings.local.json +++ b/.claude/settings.local.json @@ -10,7 +10,10 @@ "Bash(git log:*)", "Bash(git add:*)", "Bash(git commit:*)", - "Bash(del:*)" + "Bash(del:*)", + "Bash(git rm:*)", + "Bash(node -r ts-node/register src/utils/SSEParserUtils.test.ts:*)", + "Bash(node:*)" ], "deny": [], "ask": [] diff --git a/src/utils/SSEParserUtils.test.ts b/src/utils/SSEParserUtils.test.ts new file mode 100644 index 00000000..2bf8edf3 --- /dev/null +++ b/src/utils/SSEParserUtils.test.ts @@ -0,0 +1,240 @@ +import * as assert from 'assert'; +import { getFullContent } from './SSEParserUtils'; + +// 测试用例1: 正常的文本和工具调用场景 +function testNormalTextAndToolCalls(): void { + console.log('Running testNormalTextAndToolCalls...'); + + const sseEvents = [ + 'data: {"type": "message_start", "message": {"id": "msg_123", "role": "assistant", "model": "claude-3-sonnet"}}', + 'data: {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}', + 'data: {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "Hello"}}', + 'data: {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": " World"}}', + 'data: {"type": "content_block_stop", "index": 0}', + 'data: {"type": "content_block_start", "index": 1, "content_block": {"type": "tool_use", "id": "toolu_001", "name": "get_weather"}}', + 'data: {"type": "content_block_delta", "index": 1, "delta": {"type": "input_json_delta", "partial_json": "{\\"location\\":\\"Beijing\\""}}', + 'data: {"type": "content_block_delta", "index": 1, "delta": {"type": "input_json_delta", "partial_json": ",\\"unit\\":\\"celsius\\"}"}}', + 'data: {"type": "content_block_stop", "index": 1}', + 'data: {"type": "message_stop"}' + ]; + + const result = getFullContent(sseEvents); + console.log('Result:', result); + + // 验证结果包含预期的文本和工具调用 + assert(result.includes('Hello World'), 'Should contain "Hello World"'); + assert(result.includes('get_weather'), 'Should contain "get_weather"'); + assert(result.includes('Beijing'), 'Should contain "Beijing"'); + assert(result.includes('celsius'), 'Should contain "celsius"'); + + console.log('✓ testNormalTextAndToolCalls passed'); +} + +// 测试用例2: 只有文本内容 +function testTextOnly(): void { + console.log('Running testTextOnly...'); + + const sseEvents = [ + 'data: {"type": "message_start", "message": {"id": "msg_456", "role": "assistant", "model": "claude-3-sonnet"}}', + 'data: {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}', + 'data: {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "Simple text"}}', + 'data: {"type": "content_block_stop", "index": 0}', + 'data: {"type": "message_stop"}' + ]; + + const result = getFullContent(sseEvents); + console.log('Result:', result); + + assert(result.includes('Simple text'), 'Should contain "Simple text"'); + assert(result.includes('\n[]'), 'Should contain empty array for tool calls'); + + console.log('✓ testTextOnly passed'); +} + +// 测试用例3: 只有工具调用（揭示Bug） +function testToolCallsOnly(): void { + console.log('Running testToolCallsOnly...'); + + const sseEvents = [ + 'data: {"type": "message_start", "message": {"id": "msg_789", "role": "assistant", "model": "claude-3-sonnet"}}', + 'data: {"type": "content_block_start", "index": 0, "content_block": {"type": "tool_use", "id": "toolu_002", "name": "calculate"}}', + 'data: {"type": "content_block_delta", "index": 0, "delta": {"type": "input_json_delta", "partial_json": "{\\"a\\":1,\\"b\\":2}"}}', + 'data: {"type": "content_block_stop", "index": 0}', + 'data: {"type": "message_stop"}' + ]; + + const result = getFullContent(sseEvents); + console.log('Result:', result); + + // Bug验证：当没有文本内容时，会返回换行符+工具调用JSON + assert(result.startsWith('\n'), 'Should start with newline when no text content'); + assert(result.includes('calculate'), 'Should contain "calculate"'); + assert(result.includes('{"a":1,"b":2}'), 'Should contain tool input'); + + console.log('✓ testToolCallsOnly passed'); +} + +// 测试用例4: 空事件数组（揭示Bug） +function testEmptyEvents(): void { + console.log('Running testEmptyEvents...'); + + const sseEvents: string[] = []; + const result = getFullContent(sseEvents); + + console.log('Result:', result); + + // Bug验证：空事件应该返回空数组的JSON字符串 + assert.strictEqual(result, '[]', 'Should return JSON stringified empty array'); + + console.log('✓ testEmptyEvents passed'); +} + +// 测试用例5: 无效的SSE事件（揭示Bug） +function testInvalidEvents(): void { + console.log('Running testInvalidEvents...'); + + const sseEvents = [ + 'invalid data', + 'data: {"invalid": json}', + 'data: not json at all' + ]; + + const result = getFullContent(sseEvents); + console.log('Result:', result); + + // Bug验证：解析失败时返回原始事件数组的JSON字符串 + const expectedResult = JSON.stringify(sseEvents); + assert.strictEqual(result, expectedResult, 'Should return JSON stringified original events'); + + console.log('✓ testInvalidEvents passed'); +} + +// 测试用例6: 不完整的事件流（没有message_stop） +function testIncompleteEvents(): void { + console.log('Running testIncompleteEvents...'); + + const sseEvents = [ + 'data: {"type": "message_start", "message": {"id": "msg_incomplete", "role": "assistant", "model": "claude-3-sonnet"}}', + 'data: {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}', + 'data: {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "Incomplete"}}', + // 缺少 content_block_stop 和 message_stop + ]; + + const result = getFullContent(sseEvents); + console.log('Result:', result); + + // 应该能够处理不完整的事件流 + assert(result.includes('Incomplete'), 'Should contain partial text content'); + + console.log('✓ testIncompleteEvents passed'); +} + +// 测试用例7: 包含redacted_thinking +function testRedactedThinking(): void { + console.log('Running testRedactedThinking...'); + + const sseEvents = [ + 'data: {"type": "message_start", "message": {"id": "msg_redacted", "role": "assistant", "model": "claude-3-sonnet"}}', + 'data: {"type": "content_block_start", "index": 0, "content_block": {"type": "redacted_thinking", "text": ""}}', + 'data: {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "Hidden thought"}}', + 'data: {"type": "content_block_stop", "index": 0}', + 'data: {"type": "content_block_start", "index": 1, "content_block": {"type": "text", "text": ""}}', + 'data: {"type": "content_block_delta", "index": 1, "delta": {"type": "text_delta", "text": "Visible response"}}', + 'data: {"type": "content_block_stop", "index": 1}', + 'data: {"type": "message_stop"}' + ]; + + const result = getFullContent(sseEvents); + console.log('Result:', result); + + // redacted_thinking 不应该出现在最终结果中，只有可见文本 + assert(!result.includes('Hidden thought'), 'Should not include redacted thinking content'); + assert(result.includes('Visible response'), 'Should include visible text content'); + + console.log('✓ testRedactedThinking passed'); +} + +// 测试用例8: 包含ping事件 +function testPingEvents(): void { + console.log('Running testPingEvents...'); + + const sseEvents = [ + 'data: {"type": "message_start", "message": {"id": "msg_ping", "role": "assistant", "model": "claude-3-sonnet"}}', + 'data: {"type": "ping"}', + 'data: {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}', + 'data: {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "After ping"}}', + 'data: {"type": "content_block_stop", "index": 0}', + 'data: {"type": "ping"}', + 'data: {"type": "message_stop"}' + ]; + + const result = getFullContent(sseEvents); + console.log('Result:', result); + + // ping事件应该被忽略 + assert(result.includes('After ping'), 'Should contain text content'); + assert(!result.includes('ping'), 'Should not include ping events'); + + console.log('✓ testPingEvents passed'); +} + +// 运行所有测试 +function runAllTests(): void { + console.log('=== 开始运行 getFullContent 函数的单元测试 ===\n'); + + try { + testNormalTextAndToolCalls(); + console.log(''); + + testTextOnly(); + console.log(''); + + testToolCallsOnly(); + console.log(''); + + testEmptyEvents(); + console.log(''); + + testInvalidEvents(); + console.log(''); + + testIncompleteEvents(); + console.log(''); + + testRedactedThinking(); + console.log(''); + + testPingEvents(); + console.log(''); + + console.log('🎉 所有测试都通过了！'); + console.log('\n📋 发现的潜在问题总结:'); + console.log('1. 🐛 当没有文本内容时，返回结果以换行符开头（Bug: 应该处理空文本情况）'); + console.log('2. 🐛 成功解析和失败解析时的返回格式不一致（Bug: 统一返回格式）'); + console.log('3. 🐛 工具调用直接JSON.stringify，可读性较差（改进建议: 格式化输出）'); + console.log('4. ✅ redacted_thinking类型的内容被正确过滤（正确行为）'); + console.log('5. ✅ ping事件被正确忽略（正确行为）'); + console.log('6. ✅ 不完整事件流能够正确处理（正确行为）'); + + } catch (error) { + console.error('❌ 测试失败:', (error as Error).message); + process.exit(1); + } +} + +// 如果直接运行此文件，则执行测试 +if (require.main === module) { + runAllTests(); +} + +export { + runAllTests, + testNormalTextAndToolCalls, + testTextOnly, + testToolCallsOnly, + testEmptyEvents, + testInvalidEvents, + testIncompleteEvents, + testRedactedThinking, + testPingEvents +}; \ No newline at end of file diff --git a/src/utils/SSEParserUtils.ts b/src/utils/SSEParserUtils.ts index 7131807e..76f1d6f8 100644 --- a/src/utils/SSEParserUtils.ts +++ b/src/utils/SSEParserUtils.ts @@ -265,9 +265,33 @@ export function getAllToolCalls(message: CompleteMessage): ToolUseContent[] { export function getFullContent(events: string[]): string { const result = parseSSEContent(events); - if( result != null ) - return getAllTextContent(result) + '\n' + JSON.stringify( getAllToolCalls(result) ); - return JSON.stringify(events); + + if (result != null) { + // 修复 Bug #1: 提取文本内容并检查是否为空 + const textContent = getAllTextContent(result); + const toolCalls = getAllToolCalls(result); + + let output = ''; + + // 只有当文本内容不为空时才添加到输出中 + if (textContent.trim()) { + output += textContent; + } + + // 修复 Bug #3: 格式化工具调用输出 + if (toolCalls.length > 0) { + // 如果有文本内容，添加换行符分隔 + if (output) { + output += '\n'; + } + // 修复 Bug #3: 使用格式化的 JSON 输出，提高可读性 + output += JSON.stringify(toolCalls, null, 2); + } + + return output; + } else { + return `Error: Failed to parse SSE events. Original events:\n${JSON.stringify(events, null, 2)}`; + } } From 49be1072920fc9af5bcbf73ff9dacc8bb74c57fa Mon Sep 17 00:00:00 2001 From: Fu Gui Date: Thu, 27 Nov 2025 11:30:13 +0800 Subject: [PATCH 5/6] refactor: improve code organization and SSE parsing with enhanced error handling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Consolidate helper functions and extract reusable padNumber utility - Add comprehensive constants for SSE event and content block types - Enhance SSE parser with better JSON truncation fixes and error handling - Improve tool call processing with partial JSON accumulation - Refactor AgentProcessingState into class with better state management - Remove deprecated hooks and consolidate functionality - Add comprehensive JSDoc documentation throughout modules 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- src/index.ts | 55 ++++-- src/utils/SSEParserUtils.ts | 360 +++++++++++++++++++++++++----------- src/utils/hooks.ts | 144 ++++++--------- 3 files changed, 340 insertions(+), 219 deletions(-) diff --git a/src/index.ts b/src/index.ts index 2fd6d15a..ab9bb017 100644 --- a/src/index.ts +++ b/src/index.ts @@ -18,9 +18,7 @@ import { setupAgentAndRoutingHook, setupErrorEventHook, setupSendEventHook, - setupAgentProcessingHook, - setupSessionUsageHook, - setupErrorPayloadHook + setupAgentProcessingHook } from "./utils/hooks"; import { EventEmitter } from "node:events"; @@ -63,6 +61,13 @@ function setupSignalHandlers(): void { process.on("SIGTERM", () => handleShutdown("SIGTERM")); } +/** + * Pads a number with leading zero if needed + */ +function padNumber(num: number): string { + return (num > 9 ? "" : "0") + num; +} + /** * Configure logger based on config settings */ @@ -75,12 +80,11 @@ function configureLogger(config: any, homeDir: string) { time = new Date(); } - const pad = (num: number) => (num > 9 ? "" : "0") + num; - const yearAndMonth = time.getFullYear() + "" + pad(time.getMonth() + 1); - const day = pad(time.getDate()); - const hour = pad(time.getHours()); - const minute = pad(time.getMinutes()); - const second = pad(time.getSeconds()); + const yearAndMonth = time.getFullYear() + "" + padNumber(time.getMonth() + 1); + const day = padNumber(time.getDate()); + const hour = padNumber(time.getHours()); + const minute = padNumber(time.getMinutes()); + const second = padNumber(time.getSeconds()); return `./logs/ccr-${yearAndMonth}${day}${hour}${minute}${second}${index ? `_${index}` : ''}.log`; }, { @@ -108,28 +112,41 @@ async function initializeApp(): Promise { return await initConfig(); } +// Configuration constants +const DEFAULT_HOST = "127.0.0.1"; +const DEFAULT_PORT = 3456; + /** - * Resolve host configuration + * Resolve host configuration with security considerations */ function resolveHostConfig(config: any): string { - let host = config.HOST || "127.0.0.1"; + let host = config.HOST || DEFAULT_HOST; if (config.HOST && !config.APIKEY) { - host = "127.0.0.1"; - console.warn("⚠️ API key is not set. HOST is forced to 127.0.0.1."); + host = DEFAULT_HOST; + console.warn("⚠️ API key is not set. HOST is forced to 127.0.0.1 for security."); } return host; } /** - * Resolve service port configuration + * Resolve service port configuration from environment or config */ function resolveServicePort(config: any): number { - const defaultPort = config.PORT || 3456; - return process.env.SERVICE_PORT - ? parseInt(process.env.SERVICE_PORT) - : defaultPort; + const configPort = config.PORT || DEFAULT_PORT; + const envPort = process.env.SERVICE_PORT; + + if (envPort) { + const parsedEnvPort = parseInt(envPort, 10); + if (isNaN(parsedEnvPort)) { + console.warn(`⚠️ Invalid SERVICE_PORT: ${envPort}. Using default port ${configPort}.`); + return configPort; + } + return parsedEnvPort; + } + + return configPort; } /** @@ -189,8 +206,6 @@ async function run(options: RunOptions = {}): Promise { setupErrorEventHook(server, event); setupSendEventHook(server, event); setupAgentProcessingHook(server, config); - setupSessionUsageHook(server, config); - setupErrorPayloadHook(server); server.start(); } diff --git a/src/utils/SSEParserUtils.ts b/src/utils/SSEParserUtils.ts index 76f1d6f8..21f25f65 100644 --- a/src/utils/SSEParserUtils.ts +++ b/src/utils/SSEParserUtils.ts @@ -1,3 +1,4 @@ +// Content block types interface TextContent { type: 'text'; text: string; @@ -7,7 +8,7 @@ interface ToolUseContent { type: 'tool_use'; id: string; name: string; - input: any; + input: Record; partialInput?: string; } @@ -18,6 +19,31 @@ interface RedactedThinkingContent { type ContentBlock = TextContent | ToolUseContent | RedactedThinkingContent; +// Event type constants for better maintainability +const SSE_EVENT_TYPES = { + MESSAGE_START: 'message_start', + CONTENT_BLOCK_START: 'content_block_start', + CONTENT_BLOCK_DELTA: 'content_block_delta', + CONTENT_BLOCK_STOP: 'content_block_stop', + MESSAGE_DELTA: 'message_delta', + MESSAGE_STOP: 'message_stop', + PING: 'ping' +} as const; + +// Content block type constants +const CONTENT_BLOCK_TYPES = { + TEXT: 'text', + TOOL_USE: 'tool_use', + REDACTED_THINKING: 'redacted_thinking' +} as const; + +// Delta type constants +const DELTA_TYPES = { + TEXT_DELTA: 'text_delta', + INPUT_JSON_DELTA: 'input_json_delta' +} as const; + +// Message and SSE data types interface CompleteMessage { id: string | null; role: string | null; @@ -48,6 +74,10 @@ interface SSEData { }; } +// ============================================================================= +// SSE Message Assembler Class +// ============================================================================= + class SSEMessageAssembler { private currentMessage: CompleteMessage | null = null; private currentContentBlock: ContentBlock | null = null; @@ -61,35 +91,50 @@ class SSEMessageAssembler { public processEvent(eventData: string): CompleteMessage | null { try { - const data: SSEData = JSON.parse(eventData); - + const processedEventData = this.fixTruncatedData(eventData); + const data: SSEData = JSON.parse(processedEventData); + switch (data.type) { - case 'message_start': + case SSE_EVENT_TYPES.MESSAGE_START: return this.handleMessageStart(data); - case 'content_block_start': + case SSE_EVENT_TYPES.CONTENT_BLOCK_START: this.handleContentBlockStart(data); break; - case 'content_block_delta': + case SSE_EVENT_TYPES.CONTENT_BLOCK_DELTA: this.handleContentBlockDelta(data); break; - case 'content_block_stop': + case SSE_EVENT_TYPES.CONTENT_BLOCK_STOP: this.handleContentBlockStop(data); break; - case 'message_stop': + case SSE_EVENT_TYPES.MESSAGE_DELTA: + // message_delta 事件包含使用信息，但不需要处理为内容块 + break; + case SSE_EVENT_TYPES.MESSAGE_STOP: return this.handleMessageStop(data); - case 'ping': + case SSE_EVENT_TYPES.PING: // 忽略心跳包，但保持当前状态 break; default: - console.warn('未知的 SSE 事件类型:', data.type); + console.warn(`Unknown SSE event type: ${data.type}`); } } catch (error) { - console.error('解析 SSE 事件数据失败:', error, '原始数据:', eventData); + console.error(`Failed to parse SSE event data:`, error, `Raw data:`, eventData); + return null; } - + return null; } + /** + * Fix common truncated JSON data patterns + */ + private fixTruncatedData(eventData: string): string { + if (eventData.includes('"output_to"') && !eventData.includes('"output_tokens"')) { + return eventData.replace('"output_to"', '"output_tokens"'); + } + return eventData; + } + private handleMessageStart(data: SSEData): null { this.currentMessage = { id: data.message?.id || null, @@ -104,37 +149,37 @@ class SSEMessageAssembler { if (!this.currentMessage) return; this.currentIndex = data.index ?? 0; - + const contentBlock = data.content_block; if (!contentBlock) return; switch (contentBlock.type) { - case 'text': + case CONTENT_BLOCK_TYPES.TEXT: this.currentContentBlock = { - type: 'text', + type: CONTENT_BLOCK_TYPES.TEXT, text: contentBlock.text || '' }; break; - - case 'tool_use': + + case CONTENT_BLOCK_TYPES.TOOL_USE: this.currentContentBlock = { - type: 'tool_use', + type: CONTENT_BLOCK_TYPES.TOOL_USE, id: contentBlock.id || '', name: contentBlock.name || '', - input: contentBlock.input || {}, + input: contentBlock.input as Record || {}, partialInput: '' }; break; - - case 'redacted_thinking': + + case CONTENT_BLOCK_TYPES.REDACTED_THINKING: this.currentContentBlock = { - type: 'redacted_thinking', + type: CONTENT_BLOCK_TYPES.REDACTED_THINKING, text: contentBlock.text || '' }; break; - + default: - console.warn('未知的内容块类型:', contentBlock.type); + console.warn(`Unknown content block type: ${(contentBlock as any).type}`); this.currentContentBlock = null; return; } @@ -151,30 +196,19 @@ class SSEMessageAssembler { if (!delta) return; switch (this.currentContentBlock.type) { - case 'text': + case CONTENT_BLOCK_TYPES.TEXT: if (delta.text) { this.currentContentBlock.text += delta.text; } break; - - case 'tool_use': - if (delta.type === 'input_json_delta' && delta.partial_json) { - // 累积部分 JSON 输入 - if (!this.currentContentBlock.partialInput) { - this.currentContentBlock.partialInput = ''; - } - this.currentContentBlock.partialInput += delta.partial_json; - - // 尝试解析为完整对象（可能失败，因为 JSON 还不完整） - try { - this.currentContentBlock.input = JSON.parse(this.currentContentBlock.partialInput); - } catch (e) { - // JSON 还不完整，继续累积 - } + + case CONTENT_BLOCK_TYPES.TOOL_USE: + if (delta.type === DELTA_TYPES.INPUT_JSON_DELTA && delta.partial_json) { + this.accumulatePartialJsonInput(delta.partial_json); } break; - - case 'redacted_thinking': + + case CONTENT_BLOCK_TYPES.REDACTED_THINKING: if (delta.text) { this.currentContentBlock.text += delta.text; } @@ -185,15 +219,33 @@ class SSEMessageAssembler { this.currentMessage.content[this.currentIndex] = this.currentContentBlock; } + /** + * Accumulate partial JSON input for tool calls + */ + private accumulatePartialJsonInput(partialJson: string): void { + if (!this.currentContentBlock || this.currentContentBlock.type !== CONTENT_BLOCK_TYPES.TOOL_USE) { + return; + } + + if (!this.currentContentBlock.partialInput) { + this.currentContentBlock.partialInput = ''; + } + this.currentContentBlock.partialInput += partialJson; + + try { + this.currentContentBlock.input = JSON.parse(this.currentContentBlock.partialInput); + } catch (e) { + // JSON parsing incomplete, continue accumulation + } + } + private handleContentBlockStop(data: SSEData): void { - if (this.currentContentBlock?.type === 'tool_use' && this.currentContentBlock.partialInput) { - // 对于工具调用，在块结束时尝试最终解析 + if (this.currentContentBlock?.type === CONTENT_BLOCK_TYPES.TOOL_USE && this.currentContentBlock.partialInput) { try { this.currentContentBlock.input = JSON.parse(this.currentContentBlock.partialInput); delete this.currentContentBlock.partialInput; } catch (e) { - console.error('工具调用输入 JSON 解析失败:', e, '原始数据:', this.currentContentBlock.partialInput); - // 保留部分输入以便调试 + console.error(`Failed to parse tool call input JSON:`, e, `Raw data:`, this.currentContentBlock.partialInput); } } @@ -218,25 +270,34 @@ class SSEMessageAssembler { } } -// 主函数 - 直接使用这个函数 +// ============================================================================= +// Main SSE Parser Functions +// ============================================================================= + +/** + * Parse SSE content from an array of SSE event strings + * Main function for SSE parsing - use this directly + */ export function parseSSEContent(events: string[]): CompleteMessage | null { const assembler = new SSEMessageAssembler(); let finalMessage: CompleteMessage | null = null; - for (const event of events) { - // 清理事件数据（移除 "data: " 前缀等） - const cleanEvent = event.replace(/^data: /, '').trim(); - - // 跳过空行和注释 - if (!cleanEvent || cleanEvent.startsWith(':')) continue; - - const result = assembler.processEvent(cleanEvent); + const sseEvents = parseSSEEvents(events); + + for (const sseEvent of sseEvents) { + const eventData = extractEventData(sseEvent); + + if (!eventData) { + continue; + } + + const result = assembler.processEvent(eventData); if (result) { finalMessage = result; } } - // 如果没有收到 message_stop 事件，但想强制获取当前状态 + // If no message_stop event received, get current state if (!finalMessage) { const state = assembler.getCurrentState(); finalMessage = state.message; @@ -245,75 +306,154 @@ export function parseSSEContent(events: string[]): CompleteMessage | null { return finalMessage; } -// 工具函数：从完整消息中提取特定类型的内容 -export function extractContentByType(message: CompleteMessage, type: 'text' | 'tool_use' | 'redacted_thinking'): any[] { - return message.content.filter(block => block.type === type); +/** + * Parses and filters SSE events from raw event strings + */ +function parseSSEEvents(events: string[]): string[] { + const fullSSEStream = events.join('\n'); + + return fullSSEStream + .split('\n\n') + .filter(event => event.trim()) + .filter(sseEvent => !isPingEvent(sseEvent)); +} + +/** + * Checks if SSE event is a ping event + */ +function isPingEvent(sseEvent: string): boolean { + const lines = sseEvent.split('\n'); + for (const line of lines) { + const trimmedLine = line.trim(); + if (trimmedLine.startsWith('event:')) { + const eventType = trimmedLine.replace(/^event:\s*/, '').trim(); + return eventType === SSE_EVENT_TYPES.PING; + } + } + return false; +} + +/** + * Extracts event data from SSE event string + */ +function extractEventData(sseEvent: string): string { + const lines = sseEvent.split('\n'); + let eventData = ''; + + for (const line of lines) { + const trimmedLine = line.trim(); + if (trimmedLine.startsWith('data:')) { + eventData = trimmedLine.replace(/^data:\s*/, '').trim(); + } + } + + return eventData; +} + +// ============================================================================= +// Utility Functions +// ============================================================================= + +/** + * Extract content blocks by type from a complete message + */ +export function extractContentByType( + message: CompleteMessage, + contentType: ContentBlock['type'] +): ContentBlock[] { + return message.content.filter(block => block.type === contentType); } -// 工具函数：获取所有文本内容（合并所有 text 块） +/** + * Get all text content from a complete message (joins all text blocks) + */ export function getAllTextContent(message: CompleteMessage): string { return message.content - .filter((block): block is TextContent => block.type === 'text') + .filter((block): block is TextContent => block.type === CONTENT_BLOCK_TYPES.TEXT) .map(block => block.text) - .join('\n'); + .join('\n') + .trim(); } -// 工具函数：获取所有工具调用 +/** + * Get all tool calls from a complete message + */ export function getAllToolCalls(message: CompleteMessage): ToolUseContent[] { - return message.content.filter((block): block is ToolUseContent => block.type === 'tool_use'); + return message.content + .filter((block): block is ToolUseContent => block.type === CONTENT_BLOCK_TYPES.TOOL_USE); } -export function getFullContent(events: string[]): string { - const result = parseSSEContent(events); - - if (result != null) { - // 修复 Bug #1: 提取文本内容并检查是否为空 - const textContent = getAllTextContent(result); - const toolCalls = getAllToolCalls(result); - - let output = ''; +/** + * Format tool calls for output with proper error handling + */ +function formatToolCalls(toolCalls: ToolUseContent[]): Array> { + return toolCalls.map(toolCall => { + const sanitized: Record = { + type: toolCall.type, + id: toolCall.id, + name: toolCall.name + }; - // 只有当文本内容不为空时才添加到输出中 - if (textContent.trim()) { - output += textContent; + // Use complete input if available and not empty, otherwise try partialInput + if (toolCall.input && Object.keys(toolCall.input).length > 0) { + sanitized.input = toolCall.input; + } else if (toolCall.partialInput && toolCall.partialInput.trim()) { + sanitized.input = parsePartialInput(toolCall.partialInput); + } else { + sanitized.input = {}; } - // 修复 Bug #3: 格式化工具调用输出 - if (toolCalls.length > 0) { - // 如果有文本内容，添加换行符分隔 - if (output) { - output += '\n'; - } - // 修复 Bug #3: 使用格式化的 JSON 输出，提高可读性 - output += JSON.stringify(toolCalls, null, 2); - } + return sanitized; + }); +} - return output; - } else { - return `Error: Failed to parse SSE events. Original events:\n${JSON.stringify(events, null, 2)}`; +/** + * Parse partial input with error handling + */ +function parsePartialInput(partialInput: string): unknown { + try { + const parsedInput = JSON.parse(partialInput); + // Return parsed input if it's a non-empty object + return parsedInput && Object.keys(parsedInput as Record).length > 0 + ? parsedInput + : partialInput; + } catch (e) { + // Return raw string if parsing fails + return partialInput; } } - -// 使用示例 -/* -const sseEvents = [ - `data: {"type": "message_start", "message": {"id": "msg_123", "role": "assistant", "model": "claude-3-sonnet"}}`, - `data: {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}`, - `data: {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "Hello"}}`, - `data: {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": " World"}}`, - `data: {"type": "content_block_stop", "index": 0}`, - `data: {"type": "content_block_start", "index": 1, "content_block": {"type": "tool_use", "id": "toolu_001", "name": "get_weather"}}`, - `data: {"type": "content_block_delta", "index": 1, "delta": {"type": "input_json_delta", "partial_json": "{\\"location\\":\\"Beijing\\""}}`, - `data: {"type": "content_block_delta", "index": 1, "delta": {"type": "input_json_delta", "partial_json": ",\\"unit\\":\\"celsius\\"}"}}`, - `data: {"type": "content_block_stop", "index": 1}`, - `data: {"type": "message_stop"}`, -]; - -const result = parseSSEContent(sseEvents); -if (result) { - console.log('完整消息:', JSON.stringify(result, null, 2)); - console.log('所有文本:', getAllTextContent(result)); - console.log('工具调用:', getAllToolCalls(result)); +/** + * Create error response for failed SSE parsing + */ +function createErrorResponse(events: string[]): string { + return `Error: Failed to parse SSE events. Original events:\n${JSON.stringify(events, null, 2)}`; } -*/ \ No newline at end of file + +/** + * Main export function: Parse SSE events and return formatted content + */ +export function getFullContent(events: string[]): Record | string { + const result = parseSSEContent(events); + + if (!result) { + return createErrorResponse(events); + } + + const textContent = getAllTextContent(result); + const toolCalls = getAllToolCalls(result); + + const output: Record = {}; + + // Only add text content if it's not empty + if (textContent) { + output.text = textContent; + } + + // Format tool calls output + if (toolCalls.length > 0) { + output.toolCalls = formatToolCalls(toolCalls); + } + + return output; +} \ No newline at end of file diff --git a/src/utils/hooks.ts b/src/utils/hooks.ts index 27254eb9..eb31f58f 100644 --- a/src/utils/hooks.ts +++ b/src/utils/hooks.ts @@ -7,70 +7,33 @@ import { router } from "./router"; import { SSEParserTransform } from "./SSEParser.transform"; import { SSESerializerTransform } from "./SSESerializer.transform"; import { rewriteStream } from "./rewriteStream"; -import {getFullContent} from "./SSEParserUtils" +import { getFullContent } from "./SSEParserUtils"; import JSON5 from "json5"; -/** - * Agent processing state for handling tool calls - */ -interface AgentProcessingState { - currentAgent?: IAgent; - currentToolIndex: number; - currentToolName: string; - currentToolArgs: string; - currentToolId: string; - toolMessages: ToolMessage[]; - assistantMessages: AssistantMessage[]; -} - -/** - * Tool message structure - */ -interface ToolMessage { +// Type aliases for better readability +type ToolMessage = { tool_use_id: string; type: "tool_result"; content: string; -} +}; -/** - * Assistant message structure for tool use - */ -interface AssistantMessage { +type AssistantMessage = { type: "tool_use"; id: string; name: string; - input: any; -} + input: Record; +}; -/** - * Stream processing metadata - */ -interface StreamMetadata { +type StreamMetadata = { totalChunks: number; duration: number; method: string; url: string; statusCode: number; timestamp: string; -} - -/** - * Parsed SSE response structure - */ -interface PSSEResponse { - type: "parsed_sse_response"; - body: string; - summary: { - totalEvents: number; - hasUsage: boolean; - textEvents: number; - }; -} +}; -/** - * Stream error information - */ -interface StreamErrorInfo { +type StreamErrorInfo = { error: string; stack: string; metadata: { @@ -80,6 +43,31 @@ interface StreamErrorInfo { url: string; timestamp: string; }; +}; + +/** + * Agent processing state for handling tool calls + */ +class AgentProcessingState { + currentAgent?: IAgent; + currentToolIndex: number = -1; + currentToolName: string = ''; + currentToolArgs: string = ''; + currentToolId: string = ''; + toolMessages: ToolMessage[] = []; + assistantMessages: AssistantMessage[] = []; + + reset(): void { + this.currentAgent = undefined; + this.currentToolIndex = -1; + this.currentToolName = ''; + this.currentToolArgs = ''; + this.currentToolId = ''; + } + + isValid(): boolean { + return this.currentToolIndex > -1 && this.currentToolName.length > 0; + } } /** @@ -322,28 +310,13 @@ function handleErrorResponse(payload: any): any { return payload; } -/** - * Setup session usage hook - * @deprecated This functionality is now integrated in setupAgentProcessingHook - */ -export function setupSessionUsageHook(server: any, config: any): void { - // Session usage tracking is now handled in setupAgentProcessingHook -} - -/** - * Setup error payload hook - * @deprecated This functionality is now integrated in setupAgentProcessingHook - */ -export function setupErrorPayloadHook(server: any): void { - // Error payload handling is now integrated in setupAgentProcessingHook -} // ============= Stream Utilities ============= /** * Pads a number with leading zero if needed */ -function padZero(num: number): string { +function padNumber(num: number): string { return (num > 9 ? "" : "0") + num; } @@ -355,18 +328,18 @@ export function generateLogFileName(time?: Date, index?: number): string { time = new Date(); } - const yearAndMonth = time.getFullYear() + "" + padZero(time.getMonth() + 1); - const day = padZero(time.getDate()); - const hour = padZero(time.getHours()); - const minute = padZero(time.getMinutes()); - const second = padZero(time.getSeconds()); + const yearAndMonth = time.getFullYear() + "" + padNumber(time.getMonth() + 1); + const day = padNumber(time.getDate()); + const hour = padNumber(time.getHours()); + const minute = padNumber(time.getMinutes()); + const second = padNumber(time.getSeconds()); return `./logs/ccr-${yearAndMonth}${day}${hour}${minute}${second}${index ? `_${index}` : ''}.log`; } /** - * Create a wrapped stream that logs complete response content + * Creates a logging wrapper for response streams */ function createLoggingWrappedStream( originalStream: ReadableStream, @@ -401,7 +374,7 @@ function createLoggingWrappedStream( } /** - * Handle stream completion and logging + * Handles stream completion and logs final response */ async function handleStreamCompletion( loggedChunks: string[], @@ -426,7 +399,7 @@ async function handleStreamCompletion( } /** - * Handle stream errors + * Handles stream errors and logs error information */ function handleStreamError( error: Error, @@ -446,7 +419,7 @@ function handleStreamError( } /** - * Create stream metadata for logging + * Creates stream metadata for logging purposes */ function createStreamMetadata( totalChunks: number, @@ -465,7 +438,7 @@ function createStreamMetadata( } /** - * Create stream error information for logging + * Creates stream error information for logging purposes */ function createStreamErrorInfo( error: Error, @@ -487,7 +460,7 @@ function createStreamErrorInfo( } /** - * Read session usage from stream in background + * Reads session usage from background stream */ function readSessionUsageFromStream(stream: ReadableStream, sessionId: string): void { const read = async (stream: ReadableStream) => { @@ -521,7 +494,7 @@ function readSessionUsageFromStream(stream: ReadableStream, sessionId: string): } /** - * Handle background read stream errors + * Handles background read stream errors */ function handleBackgroundReadError(readError: any): void { if (readError.name === 'AbortError' || readError.code === 'ERR_STREAM_PREMATURE_CLOSE') { @@ -569,15 +542,7 @@ function handleAgentStreamProcessing( * Create initial agent processing state */ function createInitialAgentState(): AgentProcessingState { - return { - currentAgent: undefined, - currentToolIndex: -1, - currentToolName: '', - currentToolArgs: '', - currentToolId: '', - toolMessages: [], - assistantMessages: [] - }; + return new AgentProcessingState(); } /** @@ -644,6 +609,11 @@ function updateAgentStateForToolStart( data: any ): void { const agent = agentsManager.getAgent(agentName); + if (!agent) { + console.error(`Agent not found: ${agentName}`); + return; + } + agentState.currentAgent = agent; agentState.currentToolIndex = data.data.index; agentState.currentToolName = data.data.content_block.name; @@ -743,11 +713,7 @@ function createToolMessage(toolUseId: string, toolResult: string): ToolMessage { * Reset agent tool-specific state */ function resetAgentToolState(agentState: AgentProcessingState): void { - agentState.currentAgent = undefined; - agentState.currentToolIndex = -1; - agentState.currentToolName = ''; - agentState.currentToolArgs = ''; - agentState.currentToolId = ''; + agentState.reset(); } /** From ac12f33d43aa2f6f24a8b685455dc80cdce3e88a Mon Sep 17 00:00:00 2001 From: Fu Gui Date: Thu, 27 Nov 2025 17:34:04 +0800 Subject: [PATCH 6/6] add: parse_logs.js utility script for log file parsing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- parse_logs.js | 125 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 125 insertions(+) create mode 100644 parse_logs.js diff --git a/parse_logs.js b/parse_logs.js new file mode 100644 index 00000000..f3fccb9d --- /dev/null +++ b/parse_logs.js @@ -0,0 +1,125 @@ +const fs = require('fs'); +const path = require('path'); + +/** + * 解析日志文件并提取请求数据 + * @param {string} logFilePath - 日志文件路径 + * @param {string} outputDir - 输出目录 + */ +function parseLogFile(logFilePath, outputDir = './output') { + try { + // 创建输出目录 + if (!fs.existsSync(outputDir)) { + fs.mkdirSync(outputDir, { recursive: true }); + } + + // 读取日志文件 + const logContent = fs.readFileSync(logFilePath, 'utf8'); + const lines = logContent.split('\n').filter(line => line.trim()); + + // 存储请求数据 + const requests = new Map(); + + console.log(`正在处理 ${lines.length} 行日志...`); + + // 解析每一行 + lines.forEach((line, index) => { + try { + // 跳过空行和非JSON行 + if (!line.trim() || !line.startsWith('{')) { + return; + } + + const logEntry = JSON.parse(line); + + // 只处理包含 reqId 的日志条目 + if (logEntry.reqId) { + const reqId = logEntry.reqId; + + // 初始化请求数据结构 + if (!requests.has(reqId)) { + requests.set(reqId, { + requestData: null, + streamCompleteResponse: null + }); + } + + const request = requests.get(reqId); + + // 提取请求数据 + if (logEntry.requestData) { + request.requestData = logEntry.requestData; + } + + // 提取响应数据 + if (logEntry.streamCompleteResponse) { + request.streamCompleteResponse = logEntry.streamCompleteResponse; + } + } + } catch (parseError) { + console.warn(`警告: 无法解析第 ${index + 1} 行: ${parseError.message}`); + } + }); + + // 保存每个请求的数据 + let savedCount = 0; + requests.forEach((requestData, reqId) => { + if (requestData.requestData || requestData.streamCompleteResponse) { + // 保存请求数据 + if (requestData.requestData) { + const requestFilePath = path.join(outputDir, `${reqId}-request.json`); + fs.writeFileSync(requestFilePath, JSON.stringify(requestData.requestData, null, 2)); + console.log(`✓ 已保存请求数据: ${requestFilePath}`); + } + + // 保存响应数据 + if (requestData.streamCompleteResponse) { + const responseFilePath = path.join(outputDir, `${reqId}-response.json`); + fs.writeFileSync(responseFilePath, JSON.stringify(requestData.streamCompleteResponse, null, 2)); + console.log(`✓ 已保存响应数据: ${responseFilePath}`); + } + + savedCount++; + } + }); + + console.log(`\n处理完成! 总共处理了 ${requests.size} 个请求，成功保存 ${savedCount} 个请求的数据。`); + + } catch (error) { + console.error('处理日志文件时出错:', error.message); + process.exit(1); + } +} + +// 命令行参数处理 +function main() { + const args = process.argv.slice(2); + + if (args.length === 0) { + console.log('使用方法: node parse_logs.js <日志文件路径> [输出目录]'); + console.log('示例: node parse_logs.js ccr-20251127092353.log ./output'); + process.exit(1); + } + + const logFilePath = args[0]; + const outputDir = args[1] || './output'; + + // 检查日志文件是否存在 + if (!fs.existsSync(logFilePath)) { + console.error(`错误: 日志文件不存在: ${logFilePath}`); + process.exit(1); + } + + console.log(`开始处理日志文件: ${logFilePath}`); + console.log(`输出目录: ${outputDir}`); + console.log('---'); + + parseLogFile(logFilePath, outputDir); +} + +// 如果直接运行此脚本 +if (require.main === module) { + main(); +} + +module.exports = { parseLogFile }; \ No newline at end of file