/** * FileBackend - 文件存储后端 * * 将日志写入文件系统,支持: * - 按会话分目录存储 * - 按日期分文件夹 * - 自动轮转(按大小) * - JSONL 格式 */ const fs = require('fs').promises; const path = require('path'); const { createWriteStream } = require('fs'); const { BaseBackend } = require('./BaseBackend'); // 获取项目根目录 const PROJECT_ROOT = path.resolve(__dirname, '../../../../../'); class FileBackend extends BaseBackend { constructor(options = {}) { super({ name: 'file', level: 'debug', ...options }); // 默认 debug 级别,记录所有日志 // 默认使用项目根目录的 .logs/ this.baseDir = options.baseDir || path.join(PROJECT_ROOT, '.logs'); this.maxFileSize = options.maxFileSize || 100 * 1024 * 1024; // 100MB this.maxFiles = options.maxFiles || 10; this.bufferSize = options.bufferSize || 100; // 缓冲条目数 this.buffers = new Map(); // filePath -> entries[] this.streams = new Map(); // filePath -> writeStream this.fileSizes = new Map(); // filePath -> currentSize this.flushTimers = new Map(); // filePath -> timer } async init() { // 确保基础目录存在 await fs.mkdir(this.baseDir, { recursive: true }); // 启动定期刷新定时器 // unref() 防止此 timer 阻止 Node.js 进程退出 this.flushInterval = setInterval(() => { this._flushAll(); }, 1000); this.flushInterval.unref(); } /** * 计算日志文件的绝对路径 */ _getLogPath(entry) { const sessionId = entry.sessionId || 'default'; // 1. 确定目录:.logs/{sessionId}/ // 使用 path.resolve 确保路径规范化,防止路径遍历攻击 // 将 sessionId 中的非法字符替换为 _ const safeSessionId = sessionId.replace(/[^a-zA-Z0-9\-_]/g, '_'); const sessionDir = path.join(this.baseDir, safeSessionId); // 2. 确定文件名 let filename = 'orchestrator.jsonl'; // 默认 System/Orchestrator 日志 // 如果是 Agent 相关日志,且有明确的 agentName if (entry.agentName && entry.agentName !== 'orchestrator') { const agentName = entry.agentName.replace(/[^a-zA-Z0-9\-_]/g, '_'); // 尝试从 metadata 获取 iteration 信息 const iteration = entry.metadata?.iteration; if (iteration !== undefined) { // 格式: 01_agentName.jsonl const iterStr = iteration.toString().padStart(2, '0'); filename = `${iterStr}_${agentName}.jsonl`; // 如果只有 index (比如 executeWithDependencies 里的并行调用) if (entry.metadata?.index !== undefined) { filename = `${iterStr}_${agentName}_${entry.metadata.index}.jsonl`; } } else { // 没有 iteration,直接用 agentName filename = `${agentName}.jsonl`; } } return path.join(sessionDir, filename); } async write(entry) { if (!this._shouldLog(entry)) return; // 获取目标文件路径 const filePath = this._getLogPath(entry); // 初始化缓冲区 (Key 为文件路径) if (!this.buffers.has(filePath)) { this.buffers.set(filePath, []); } const buffer = this.buffers.get(filePath); buffer.push(this._formatEntry(entry)); // 缓冲满时刷新 if (buffer.length >= this.bufferSize) { await this._flush(filePath); } else { // 设置延迟刷新 this._scheduleFlush(filePath); } } async query(filters = {}) { const { sessionId, startTime, endTime, category, agentName, limit = 100 } = filters; const results = []; try { // 如果没有指定 sessionId,搜索所有会话 const sessionsToSearch = sessionId ? [sessionId] : await this._getAllSessionIds(); // 这里可能会有性能问题,因为现在每个 Session 有多个文件 // 简单起见,我们优先读取 orchestrator.jsonl,如果用户查特定 agent 再读特定文件 // 但为了保证时间线,理想情况是读取该 Session 下所有文件并 Merge Sort // 本次重构先实现合并读取。 for (const sid of sessionsToSearch) { const entries = await this._readSessionLogs(sid); // 读取全部文件 for (const entry of entries) { // 应用过滤器 if (startTime && new Date(entry.timestamp) < new Date(startTime)) continue; if (endTime && new Date(entry.timestamp) > new Date(endTime)) continue; if (category && entry.category !== category) continue; if (agentName && entry.agentName !== agentName) continue; results.push(entry); // 注意:这里 limit 是针对全局的,为了更精确,应该全部读完再 sort 再 slice // 暂不 break,除非为了性能 } } // 按时间戳全量排序 return results.sort((a, b) => new Date(a.timestamp) - new Date(b.timestamp)).slice(0, limit); } catch (err) { console.error('[FileBackend] Query error:', err); return []; } } async close() { // 清除定时器 if (this.flushInterval) { clearInterval(this.flushInterval); } // 取消所有延迟刷新 for (const timer of this.flushTimers.values()) { clearTimeout(timer); } this.flushTimers.clear(); // 刷新所有缓冲区 await this._flushAll(); // 关闭所有流 for (const [filePath, stream] of this.streams) { stream.end(); } this.streams.clear(); } /** * 刷新指定文件的缓冲区 * @param {string} filePath - 完整文件路径 */ async _flush(filePath) { const buffer = this.buffers.get(filePath); if (!buffer || buffer.length === 0) return; // 清空缓冲区 this.buffers.set(filePath, []); clearTimeout(this.flushTimers.get(filePath)); this.flushTimers.delete(filePath); try { // 确保目录存在 const dir = path.dirname(filePath); await fs.mkdir(dir, { recursive: true }); // 获取或创建写入流 let stream = this.streams.get(filePath); let currentSize = this.fileSizes.get(filePath) || 0; // 检查是否需要轮转 if (currentSize > this.maxFileSize) { await this._rotate(filePath); stream = null; currentSize = 0; } if (!stream) { stream = createWriteStream(filePath, { flags: 'a' }); this.streams.set(filePath, stream); // 获取当前文件大小(仅首次打开时) if (!this.fileSizes.has(filePath)) { try { const stats = await fs.stat(filePath); currentSize = stats.size; } catch { currentSize = 0; } } } // 写入数据 const data = buffer.map(e => JSON.stringify(e)).join('\n') + '\n'; stream.write(data); // 更新大小 currentSize += Buffer.byteLength(data, 'utf8'); this.fileSizes.set(filePath, currentSize); } catch (err) { console.error(`[FileBackend] Failed to flush to ${filePath}:`, err); } } /** * 刷新所有缓冲区 * @private */ async _flushAll() { const promises = []; for (const filePath of this.buffers.keys()) { promises.push(this._flush(filePath)); } await Promise.all(promises); } /** * 设置延迟刷新 * @param {string} filePath */ _scheduleFlush(filePath) { if (this.flushTimers.has(filePath)) return; const timer = setTimeout(() => { this._flush(filePath); }, 100); // 100ms 延迟 timer.unref(); this.flushTimers.set(filePath, timer); } /** * 轮转日志文件 * @param {string} filePath */ async _rotate(filePath) { const stream = this.streams.get(filePath); if (stream) { stream.end(); this.streams.delete(filePath); } const rotatedFile = filePath.replace('.jsonl', `.jsonl.1`); try { await fs.rename(filePath, rotatedFile); } catch (err) { console.error(`[FileBackend] Failed to rotate log for ${filePath}:`, err); } this.fileSizes.set(filePath, 0); } /** * 获取所有会话ID * 扫描 .logs/ 下的所有子目录 * @private */ async _getAllSessionIds() { try { const items = await fs.readdir(this.baseDir, { withFileTypes: true }); return items .filter(item => item.isDirectory()) .map(item => item.name); } catch { return []; } } /** * 读取会话日志(合并目录下所有文件) * @private */ async _readSessionLogs(sessionId) { const sessionDir = path.join(this.baseDir, sessionId); const entries = []; try { const files = await fs.readdir(sessionDir); const jsonlFiles = files.filter(f => f.endsWith('.jsonl') && !f.includes('.jsonl.')); for (const file of jsonlFiles) { const content = await fs.readFile(path.join(sessionDir, file), 'utf8'); const lines = content.trim().split('\n').filter(line => line); for (const line of lines) { try { entries.push(JSON.parse(line)); } catch { // 忽略 } } } } catch { // 目录不存在 } return entries; } } module.exports = { FileBackend };