多格式文件解析：JSONL / SQLite / Event Stream

本文面向：需要处理多种数据格式解析的开发者。
预计阅读时间：12 分钟
最终效果：理解 JSONL 流式解析、SQLite WASM 解析、Event Stream 重建三种策略，以及 SourceAdapter 统一接口背后的设计思路。

问题背景

Claude Code、Cursor、Codex CLI、Trae、GitHub Copilot —— 这 5 个工具各自把对话记录存在不同地方、用不同格式：

如果为每个工具写一套独立的解析脚本，代码会迅速失控。我们需要一个统一的插件接口。

统一接口：SourceAdapter

所有适配器实现同一个 TypeScript 接口：

interfaceSourceAdapter{readonlyname:string;// 唯一标识，如 'claude-code'readonlydisplayName:string;// UI 显示名，如 'Claude Code'readonlyparserVersion?:string;// 解析器版本，用于远程导入去重detect():Promise<SourceInfo|null>;// 当前机器是否有这个数据源scan():Promise<ConversationMeta[]>;// 扫描所有会话文件（只拿元数据）parse(meta:ConversationMeta):Promise<ParsedConversation>;// 解析单个会话}

三层职责分离：detect()判断数据源是否存在，scan()快速遍历文件列表（不读内容），parse()才真正解析文件。这样 scan 阶段可以做到很快，配合file_size+file_mtime的去重策略，跳过没有变化的文件。

格式一：JSONL 流式读取

JSONL（JSON Lines）是最常见的格式，每行一个独立的 JSON 对象。Claude Code 和 Copilot 都用这种格式，但内部结构差异很大。

readline 流式处理

Node.js 的readline模块配合createReadStream可以逐行读取文件，内存占用恒定，不会因为文件大而 OOM：

import{createReadStream}from'node:fs';import{createInterface}from'node:readline';constfileStream=createReadStream(meta.filePath,{encoding:'utf-8'});constrl=createInterface({input:fileStream,crlfDelay:Number.POSITIVE_INFINITY,// 处理 \r\n 换行});forawait(constlineofrl){if(!line.trim())continue;letparsed;try{parsed=JSON.parse(line);}catch{continue;// 跳过格式错误的行}// ... 处理 parsed}

关键细节：crlfDelay: Infinity确保 Windows 的\r\n换行被正确处理，不会把一行拆成两行。

Claude Code 的噪音过滤

Claude Code 的 JSONL 里混杂了大量中间状态：streaming delta、tool progress、file snapshots 等。直接导入会得到几十倍的噪音数据。

过滤逻辑用一个Set做快速查找：

constSKIP_TYPES=newSet(['file-history-snapshot','last-prompt','progress','agent_progress','hook_progress','queue-operation','message',// streaming delta'tool_use',// streaming tool delta'tool_result','thinking','text','tool_reference',]);functionisRelevantMessage(line:RawMessage):boolean{if(!line.uuid)returnfalse;// 流式 delta 没有 uuidif(line.type&&SKIP_TYPES.has(line.type))returnfalse;if(line.type==='system')returnfalse;// 全部系统消息跳过return['user','assistant'].includes(line.type??'');}

保留 uuid 是第一个筛选条件 —— 流式增量片段不会带 uuid，有 uuid 的才是完整消息。然后排除已知噪音类型，最后只保留user和assistant两种角色。

Claude Code 的消息内容里还会嵌入系统标签，需要额外清洗：

functionsanitizeContent(text:string):string{letresult=text;result=result.replace(/<system-reminder>[\s\S]*?<\/system-reminder>/g,'');result=result.replace(/<command-name>[^<]*<\/command-name>/g,'');result=result.replace(/<command-message>[^<]*<\/command-message>/g,'');// ... 更多标签returnresult.trim();}

Copilot 的双层 JSONL

Copilot 的 JSONL 文件结构不一样。每行是一个CopilotSessionSnapshot，但关键数据在kind: 0的首行快照里，后续行是 UI 状态补丁（kind: 1），不需要解析。

所以 Copilot 适配器只读第一行：

// 只读第一行快照，忽略后续 UI 补丁constsnapshotText=awaitreadFirstLine(meta.filePath);

readFirstLine的实现也很简洁 —— 拿到第一行后立即关闭流：

asyncfunctionreadFirstLine(filePath:string):Promise<string|null>{conststream=createReadStream(filePath,{encoding:'utf-8'});constrl=createInterface({input:stream,crlfDelay:Infinity});forawait(constlineofrl){rl.close();stream.destroy();returnline;}returnnull;}

快照里的数据结构是请求-响应配对的，一个requests数组里每个元素包含message（用户输入）和response（助手回复，是一个响应项数组）。响应项按kind字段区分类型：text是正文，thinking是思考过程，toolInvocationSerialized标记工具调用。

格式二：SQLite WASM 解析

Cursor 和 Trae 把对话存在 VS Code 的state.vscdb文件里。这是一个标准的 SQLite 数据库，但有一个问题：VS Code 运行时会锁定文件。

sql.js 方案

我们用 sql.js —— 一个编译为 WASM 的 SQLite 实现 —— 来读取数据库。核心思路是把整个 .db 文件读进内存，然后用 sql.js 打开内存中的副本，完全绕过文件锁：

import{readFileSync}from'node:fs';importinitSqlJsfrom'sql.js';letsqlJsInstance=null;asyncfunctiongetSqlJs(){if(!sqlJsInstance){sqlJsInstance=awaitinitSqlJs();}returnsqlJsInstance;}asyncfunctionopenVscdb(dbPath:string){try{constSQL=awaitgetSqlJs();constbuf=readFileSync(dbPath);// 读进内存returnnewSQL.Database(buf);// 内存中打开}catch{// 文件锁定或损坏 —— 等 500ms 重试一次awaitnewPromise(r=>setTimeout(r,500));constSQL=awaitgetSqlJs();constbuf=readFileSync(dbPath);returnnewSQL.Database(buf);}}

重试机制很重要：VS Code 在写入数据库时会短暂锁定文件，一次重试基本能解决问题。

Cursor：ItemTable + cursorDiskKV

Cursor 的数据分布在两个地方：

1. 工作区级state.vscdb——ItemTable里存着composer.composerData，记录该工作区所有 composer 会话的 ID 列表
2. 全局state.vscdb——cursorDiskKV表里存着每个 bubble（消息气泡）的实际内容

解析流程是先从工作区 DB 获取 composer ID 列表，再从全局 DB 的cursorDiskKV表按bubbleId:<composerId>:*的 pattern 查询所有气泡：

constresult=db.exec(`SELECT [key], value FROM cursorDiskKV WHERE [key] LIKE 'bubbleId:${composerId}:%'`);for(constrowofresult[0].values){constbubble=JSON.parse(row[1]asstring);// 检查 schema 版本if(bubble._v&&bubble._v>3){console.warn(`[Cursor] Unknown bubble schema version:${bubble._v}`);}constmsgType=bubble.type===1?'user':'assistant';// ...}

这里有一个防御性检查：bubble._v > 3时打 warning。Cursor 的数据格式会随版本迭代变化，当遇到未知的 schema 版本时，我们记录日志而不是直接崩溃，保证向前兼容。

还有一个边界情况：某些 composer 有气泡数据但没有对应的工作区条目（比如工作区被删除了）。我们通过扫描全局 DB 里的bubbleId:*前缀来发现这些"孤儿"会话。

Trae：单一 KV 存储

Trae 比 Cursor 简单。它用一个 key 为memento/icube-ai-agent-storage的 KV 条目存下所有会话数据：

constresult=db.exec("SELECT value FROM ItemTable WHERE [key] = 'memento/icube-ai-agent-storage'");constdata=JSON.parse(result[0].values[0][0]asstring);

返回的 JSON 里有一个list数组，每个元素是一个会话，包含messages数组。注意 value 可能是字符串也可能是Uint8Array（sql.js 对 blob 和 text 的处理），需要做类型判断：

constraw=result[0].values[0][0];conststr=typeofraw==='string'?raw:Buffer.from(rawasUint8Array).toString('utf8');

Trae 的助手消息存储也比较特殊。真正的回复内容可能在agentTaskContent里，而不是content字段。需要递归提取proposal、guideline.planItems中finish步骤的thought等。

格式三：Event Stream 重建

Codex CLI 的 JSONL 文件不是简单的消息列表，而是一个带类型的事件流。每行是一个事件，包含type和payload：

{"type":"session_meta","payload":{"id":"...","cwd":"/path"}}{"type":"event_msg","payload":{"type":"user_message","message":"..."}}{"type":"event_msg","payload":{"type":"agent_message","message":"..."}}{"type":"response_item","payload":{"type":"message","role":"assistant","content":[...]}}{"type":"response_item","payload":{"type":"function_call","name":"shell"}}

关键挑战是同一个助手回复可能有多个事件表示：event_msg/agent_message是纯文本版本，response_item/message是结构化版本（含 content blocks），两者内容相同但格式不同。

去重策略是记住最后一条助手消息的内容，遇到新消息时对比：

letlastAssistantMsg=null;// event_msg/agent_message 处理if(parsed.type==='event_msg'&&payloadType==='agent_message'){constmsg={content:text,/* ... */};messages.push(msg);lastAssistantMsg=msg;}// response_item/message 处理if(parsed.type==='response_item'&&payloadType==='message'){constfullText=/* 提取 output_text blocks */;if(lastAssistantMsg&&lastAssistantMsg.content===fullText){continue;// 跳过重复}// ...}

工具调用的标记也有意思。response_item/function_call事件本身不产生消息，但需要回溯标记前一条助手消息的hasToolUse = true：

if(payloadType==='function_call'||payloadType==='custom_tool_call'){if(lastAssistantMsg){lastAssistantMsg.hasToolUse=true;}}

Codex CLI 还有一个session_index.jsonl文件存储会话的显示名称（slug），解析时需要额外加载这个索引来补充元数据。

去重与增量导入

解析只是第一步。导入时还需要高效的去重策略，避免重复解析没变化的文件。

ChatCrystal 的去重基于(id, source)的复合主键加上file_size+file_mtime的变更检测：

constexisting=db.exec("SELECT file_size, file_mtime FROM conversations WHERE id = ? AND source = ?",[meta.id,meta.source]);if(existing.length>0){const[existingSize,existingMtime]=existing[0].values[0];if(Number(existingSize)===meta.fileSize&&existingMtime===meta.fileMtime){progress.skipped++;continue;// 文件没变，跳过}}

这个策略的好处是 scan 阶段只需要stat()获取文件大小和修改时间，不需要读文件内容。对于包含数千个会话文件的目录，这能把 scan 时间从分钟级降到秒级。

对于 SQLite 数据源（Cursor、Trae），情况比较特殊 —— 多个会话共享同一个 .db 文件。这意味着文件的 mtime 会因为任何会话的更新而变化。我们的做法是用会话自身的创建时间（createdAt）作为fileMtime，而不是数据库文件的修改时间。

错误处理策略

面对真实用户数据，各种异常都会出现。每个适配器都遵循相同的原则：不因单个文件的错误阻断整个导入流程。

JSONL 解析中，JSON.parse的异常被 catch 后直接continue跳过该行。SQLite 打开失败会重试一次。工作区目录损坏被跳过。每个错误都会写入import_log表，导入完成后可以在日志中查看。

try{constparsed=awaitadapter.parse(meta);// ... 正常处理}catch(err){progress.errors++;db.run(`INSERT INTO import_log (file_path, status, message) VALUES (?, 'error', ?)`,[meta.filePath,err.message]);}

消息数不足 2 条的会话也会被跳过 —— 一条消息的会话通常没有总结价值。

设计复盘

回顾整个解析系统，几个设计决策经受住了时间考验：

插件注册机制让新增数据源只需要写一个文件。注册发生在模块加载时，一行registerAdapter(copilotAdapter)就够了。

scan/parse 分离让变更检测变得廉价。scan 只拿文件列表和 stat 信息，parse 才真正读内容。配合 file_size + file_mtime 的去重，大量文件可以秒级跳过。

内存中的 SQLite 副本绕过了 VS Code 的文件锁问题，代价是一次性把整个 .db 文件读进内存。对于通常只有几十 MB 的 vscdb 文件来说完全可以接受。

向前兼容的 schema 检查（如 Cursor 的_v版本号）让我们在工具升级格式后不会立即崩溃，而是打 warning 并尽力解析。

这套系统目前覆盖了 5 个数据源、3 种文件格式，新增一个数据源的典型开发时间是半天。如果你也在做类似的多源数据聚合，希望这些解析策略对你有帮助。

项目地址：github.com/ZengLiangYi/ChatCrystal

如有疑问欢迎在 GitHub Issues 或私信交流，很乐意解答。