Claude Code jsonl → Langfuse Watcher

独立守护进程，轮询 Claude Code session jsonl，将对话增量上报到 Langfuse。无需 Hook，不依赖 Claude Code 进程内事件。

方案对比

维度	Hook 方案	Watcher 方案（本目录）
触发方式	Claude Code `Stop` 等 Hook 事件	独立进程轮询 jsonl 文件
实时性	会话结束或特定事件后批量上报	默认流式：边写 jsonl 边推 Langfuse
部署	需配置 `.claude/hooks`	单独启动 `run_watcher.sh` 即可
依赖	Hook 脚本 + Langfuse 客户端	单文件 `claude_langfuse_watcher.py`（自包含）
断点续传	依赖 Hook 状态	`offset` + `turn_count` 持久化到 state.json

架构设计

总体架构

模块分层

层级	模块	职责
调度层	`JsonlWatcher`	多文件跟踪、轮询、`rescan` 发现新 session、idle 超时
读取层	`read_new_jsonl` / `SessionState`	基于 `offset` + `buffer` 的 tail-f 增量读，半行缓冲
解析层	`build_turns` / `Turn`	jsonl 消息流 → 对话轮次，tool_result 折叠进当前 turn
上报层	`_emit_turn` / `process_turn_streaming`	Turn → Langfuse trace/generation/span 树
子 Agent	`discover_subagent_files` / `match_subagent`	关联 `agent-*.jsonl`，嵌套在 `Task`/`Agent` span 下
持久化	`FileLock` / `SingleInstanceLock` / `state_key`	并发安全、断点续传、单实例

Langfuse 对象树（单 Turn）

trace: "Claude Code - Turn N"          [session_id = Claude session UUID]
│
├── generation: "Claude Response"      ← 按 message.id 分组，每次 API 调用一条
├── span: "Tool: Read"                 ← 每个 tool_use 一条，按 start_time 与 generation 交错
│   └── span: "Subagent[Explore] Turn 1"   ← Task/Agent 下嵌套（可选）
│       ├── generation ...
│       └── span: "Tool: Bash" ...
├── generation: "Claude Response 2"    ← 工具结果后的续接调用
│
└── trace.output = 本轮最终用户可见回答

核心实现思路

1. 增量读取（tail-f + 半行缓冲）

Claude Code 追加写 jsonl 时，最后一行可能未写完。watcher 用 offset 记录已读字节位置，buffer 暂存不完整行，下一轮拼接后再解析。

字节进度（offset / buffer）：每轮 read_new_jsonl 都更新
业务进度（turn_count）：仅 Langfuse flush 成功后才 +1，避免「读了但未上报」导致丢数据

文件被截断/重建时（size < offset），自动重置从头读。

2. Turn 组装

以真人 user 消息为边界切分轮次：

tool_result（user 角色）→ 归入当前 turn，不新开
Skill 注入等合成 user（如 "Base directory for this skill:"）→ 不拆 turn
permission-mode、attachment、system 等元数据行 → 跳过

完成判定：最后一条 assistant 的 stop_reason == "end_turn"。

3. 流式 vs 批量上报

模式	开关	行为
流式（默认）	`CC_LANGFUSE_STREAMING=true`	user 到达即建 trace；generation 随文本增长多次 `update`；`tool_use` 先开 span，`tool_result` 后 `end`
批量	`CC_LANGFUSE_STREAMING=false`	turn 完整结束后一次性 `emit_main_turn`

流式模式将 trace_id、generation/span id 存入 streaming 字段，写入 state.json，支持进程重启后续传。

4. 子 Agent 关联

当工具名为 Task 或 Agent 时，扫描 subagents/agent-*.jsonl 并嵌套上报。

当前采用启发式匹配（非 toolUseResult.agentId 精确关联）：

匹配依据：subagent_type + 时间差 + used_subagents 去重
时间窗口：CC_LANGFUSE_SUBAGENT_TIME_WINDOW_S（默认 10s）
流式模式：Task span 未结束即开始 tail 子 Agent jsonl

日志中已有 toolUseResult.agentId 可做精确关联，详见 QA_claude_langfuse_watcher.md Q5–Q8。

5. 并发与单实例

SingleInstanceLock：同 $HOME 只允许一个 watcher 进程
FileLock：保护 langfuse_watcher_state.json 读-改-写
find_other_watcher_pids()：扫 /proc 双重检查遗留进程

目录与文件

文件	说明
`claude_langfuse_watcher.py`	主程序（解析 + 上报 + 调度，单文件自包含）
`run_watcher.sh`	常驻监听推荐入口
`e2e_langfuse_watcher_test.py`	端到端测试
`run_e2e_langfuse_watcher_test.sh`	测试脚本入口
`QA_claude_langfuse_watcher.md`	代码细节 Q&A（本次阅读笔记）

运行时产生的状态文件

路径：~/.claude/state/

文件	说明
`langfuse_watcher_state.json`	每个 session 的 `offset`、`turn_count`、`streaming`
`langfuse_watcher.log`	运行审计日志
`langfuse_watcher_instance.lock`	单实例锁（持锁 PID）
`langfuse_watcher_state.lock`	state.json 读写互斥锁

使用方式

前置条件

bash

pip install -r ../requirements.txt   # 需要 langfuse>=2,<3

环境变量（必填）

bash

export TRACE_TO_LANGFUSE=true
export LANGFUSE_PUBLIC_KEY=pk-lf-xxx      # 或 CC_LANGFUSE_PUBLIC_KEY
export LANGFUSE_SECRET_KEY=sk-lf-xxx      # 或 CC_LANGFUSE_SECRET_KEY
export LANGFUSE_BASE_URL=https://your-langfuse-host   # 或 CC_LANGFUSE_BASE_URL

推荐：常驻监听（`run_watcher.sh`）

bash

# 编辑 run_watcher.sh 中的 Langfuse 凭证与 WATCH_DIRS
./run_watcher.sh

run_watcher.sh 默认行为：

CC_LANGFUSE_STREAMING=true（流式上报）
CC_LANGFUSE_WATCH_IDLE_S=0（永不因空闲停止跟踪）
通过 WATCH_DIRS（冒号分隔）或 PROJECTS_ROOT（单目录）指定监听路径

bash

# 自定义监听目录
WATCH_DIRS="$HOME/.claude/projects:$HOME/.trpc-claudecode/projects" ./run_watcher.sh

# 传递额外参数给 Python
./run_watcher.sh --poll-interval 2.0

直接调用 Python

监听 projects 目录（自动发现新 session）：

bash

python3 claude_langfuse_watcher.py \
  --watch-dir ~/.claude/projects \
  --watch-dir ~/.trpc-claudecode/projects

监听单个 transcript（开发/调试）：

bash

python3 claude_langfuse_watcher.py \
  --transcript ~/.claude/projects/<project>/<session-uuid>.jsonl

文件尚未创建时也会注册等待（wait_for_create），出现后自动开始跟踪。

单次处理（CI / 测试）：

bash

python3 claude_langfuse_watcher.py \
  --transcript demo.jsonl \
  --once \
  --no-watch

关闭流式，改用批量模式：

bash

CC_LANGFUSE_STREAMING=false python3 claude_langfuse_watcher.py --watch-dir ~/.claude/projects

CLI 参数

参数	默认	说明
`--transcript PATH`	-	监听单个 jsonl；可与 `--watch-dir` 同时使用
`--watch-dir DIR`	`~/.claude/projects`	递归发现主 session jsonl；可多次指定
`--no-watch`	false	不扫描目录，仅用 `--transcript`
`--poll-interval SEC`	`1.0`	轮询间隔（`CC_LANGFUSE_WATCH_POLL_S`）
`--idle-timeout SEC`	`300`	文件无新内容多久后停止跟踪；`0`=永不（`CC_LANGFUSE_WATCH_IDLE_S`）
`--once`	false	只跑一轮后退出
`--task-id ID`	-	写入每个 trace metadata 的 `task_id`

启动时目录为空不会退出（--once 除外），会持续 rescan 等待新 session。

业务 task_id

用于在 Langfuse 中串联多个 session / trace：

bash

export CC_LANGFUSE_TASK_ID=my_feature_123
# 或
python3 claude_langfuse_watcher.py --watch-dir ... --task-id my_feature_123

查看运行日志

bash

tail -f ~/.claude/state/langfuse_watcher.log

调试细节：

bash

CC_LANGFUSE_DEBUG=true ./run_watcher.sh

环境变量速查

变量	默认	说明
`TRACE_TO_LANGFUSE`	-	必须为 `true` 才启动上报
`LANGFUSE_PUBLIC_KEY` / `CC_LANGFUSE_PUBLIC_KEY`	-	Langfuse 公钥
`LANGFUSE_SECRET_KEY` / `CC_LANGFUSE_SECRET_KEY`	-	Langfuse 私钥
`LANGFUSE_BASE_URL` / `CC_LANGFUSE_BASE_URL`	`https://cloud.langfuse.com`	Langfuse 服务地址
`CC_LANGFUSE_STREAMING`	`true`	流式实时上报
`CC_LANGFUSE_WATCH_POLL_S`	`1.0`	轮询间隔（秒）
`CC_LANGFUSE_WATCH_IDLE_S`	`300`	空闲超时（秒）；`0`=永不
`CC_LANGFUSE_TASK_ID`	-	业务 task_id
`CC_LANGFUSE_DEBUG`	`false`	开启 DEBUG 日志
`CC_LANGFUSE_MAX_CHARS`	`20000`	单字段上报最大字符数
`CC_LANGFUSE_REDACT`	`true`	上报前脱敏 API key 等
`CC_LANGFUSE_INCLUDE_SUBAGENTS`	`true`	是否展开子 Agent
`CC_LANGFUSE_SUBAGENT_TIME_WINDOW_S`	`10`	子 Agent 启发式匹配时间窗口

jsonl 与目录布局

主 session jsonl

文件名：<session-uuid>.jsonl（标准 UUID 格式）
常见路径：
- 官方 Claude Code：~/.claude/projects/<project-hash>/<uuid>.jsonl
- trpc-claudecode：~/.trpc-claudecode/projects/...

子 Agent jsonl

文件名：agent-<agentId>.jsonl
常见路径：
- Layout B：.../<session-uuid>/subagents/agent-*.jsonl
- Layout A：与主 jsonl 同目录或 <stem>/subagents/

watcher 只跟踪主 session jsonl（is_main_session_jsonl），不直接跟踪 agent-*.jsonl；子 Agent 由主 session 中的 Task/Agent 工具调用触发关联。

数据流（单轮轮询）

run_once()
  │
  ├─ rescan()                     发现新 <uuid>.jsonl
  │
  ├─ FileLock → load state.json
  │
  └─ 对每个 TrackedFile:
        read_new_jsonl(offset)    增量读新行
        build_turns()             组装 Turn
        process_file()            流式或批量上报
        langfuse.flush()          成功则 turn_count++
        save state.json

测试

bash

./run_e2e_langfuse_watcher_test.sh

或手动对 demo transcript：

bash

export TRACE_TO_LANGFUSE=true
export LANGFUSE_PUBLIC_KEY=...
export LANGFUSE_SECRET_KEY=...
export LANGFUSE_BASE_URL=...

python3 claude_langfuse_watcher.py \
  --transcript ../transcript_demo/c39794f4-2660-4ee0-810b-91991a3f7f3f.jsonl \
  --once --no-watch

配合 ../transcript_demo/generate_streaming_transcript.py 可模拟流式写入。

常见问题

启动报 "Another claude_langfuse_watcher is already running"

同 $HOME 下已有 watcher 进程。查看占锁 PID：

bash

cat ~/.claude/state/langfuse_watcher_instance.lock
# 或
ps aux | grep claude_langfuse_watcher

停止旧进程后再启动。

Langfuse 上看不到 trace

确认 TRACE_TO_LANGFUSE=true
检查凭证与 LANGFUSE_BASE_URL
查看 ~/.claude/state/langfuse_watcher.log 是否有 emitted / streaming trace started
确认 jsonl 路径被 watcher 跟踪（discovered new transcript 日志）

子 Agent 内容缺失或挂错

当前为启发式匹配，并行多个同类型子 Agent 时可能关联错误。参见 QA_claude_langfuse_watcher.md。

上报重复或漏报

重复：通常不应发生；检查是否启动了多个 watcher
漏报：查看 partial emit debug 日志；可能是 langfuse.flush 失败，turn_count 未推进，下次会重试

code index

hooks

langfuse trace

docs

hook

watcher

Claude Code jsonl → Langfuse Watcher

方案对比

架构设计

总体架构

模块分层

Langfuse 对象树（单 Turn）

核心实现思路

1. 增量读取（tail-f + 半行缓冲）

2. Turn 组装

3. 流式 vs 批量上报

4. 子 Agent 关联

5. 并发与单实例

目录与文件

运行时产生的状态文件

使用方式

前置条件

环境变量（必填）

推荐：常驻监听（`run_watcher.sh`）

直接调用 Python

CLI 参数

业务 task_id

查看运行日志

环境变量速查

jsonl 与目录布局

主 session jsonl

子 Agent jsonl

数据流（单轮轮询）

测试

常见问题

启动报 "Another claude_langfuse_watcher is already running"

Langfuse 上看不到 trace

子 Agent 内容缺失或挂错

上报重复或漏报

延伸阅读

docs

hook

watcher

Claude Code jsonl → Langfuse Watcher ​

方案对比 ​

架构设计 ​

总体架构 ​

模块分层 ​

Langfuse 对象树（单 Turn） ​

核心实现思路 ​

1. 增量读取（tail-f + 半行缓冲） ​

2. Turn 组装 ​

3. 流式 vs 批量上报 ​

4. 子 Agent 关联 ​

5. 并发与单实例 ​

目录与文件 ​

运行时产生的状态文件 ​

使用方式 ​

前置条件 ​

环境变量（必填） ​

推荐：常驻监听（run_watcher.sh） ​

直接调用 Python ​

CLI 参数 ​

业务 task_id ​

查看运行日志 ​

环境变量速查 ​

jsonl 与目录布局 ​

主 session jsonl ​

子 Agent jsonl ​

数据流（单轮轮询） ​

测试 ​

常见问题 ​

启动报 "Another claude_langfuse_watcher is already running" ​

Langfuse 上看不到 trace ​

子 Agent 内容缺失或挂错 ​

上报重复或漏报 ​

延伸阅读 ​

Claude Code jsonl → Langfuse Watcher

方案对比

架构设计

总体架构

模块分层

Langfuse 对象树（单 Turn）

核心实现思路

1. 增量读取（tail-f + 半行缓冲）

2. Turn 组装

3. 流式 vs 批量上报

4. 子 Agent 关联

5. 并发与单实例

目录与文件

运行时产生的状态文件

使用方式

前置条件

环境变量（必填）

推荐：常驻监听（`run_watcher.sh`）

直接调用 Python

CLI 参数

业务 task_id

查看运行日志

环境变量速查

jsonl 与目录布局

主 session jsonl

子 Agent jsonl

数据流（单轮轮询）

测试

常见问题

启动报 "Another claude_langfuse_watcher is already running"

Langfuse 上看不到 trace

子 Agent 内容缺失或挂错

上报重复或漏报

延伸阅读