docs: add subagent guidance and replay diff skill

frontend/app/app/doc/[slug]/page.tsx

@@ -56,6 +56,11 @@ const technicalDocs = {
 		file: "10-memory-research.md",
 		image: null,
 	},
+	"11-subagents": {
+		title: "子 Agent 使用边界",
+		file: "11-subagents.md",
+		image: null,
+	},
 } as const;
 
 type TechnicalDocSlug = keyof typeof technicalDocs;

frontend/app/app/doc/page.tsx

@@ -123,6 +123,27 @@ const sections: Section[] = [
 		],
 		decision: "多人使用时，隔离、恢复、取消、观测和产物管理比单次回答更重要。",
 	},
+	{
+		id: "subagents",
+		no: "11",
+		title: "子 Agent 边界",
+		summary:
+			"子 Agent 是主 Agent 派出的隔离工作单元，适合长耗时、可并行、读多写少、边界清晰的子任务。",
+		doc: "/doc/11-subagents",
+		points: [
+			"默认仍然是主 Agent + Skill + Tool；子 Agent 不是复杂任务的默认答案。",
+			"适合多个 rid、多个文件、多个候选方向并行探索，或让 verification 独立验收。",
+			"不适合强共享上下文、强顺序依赖、多个 Agent 同时写同一份最终产物的任务。",
+		],
+		code: [
+			"src/agent_tools.py",
+			"src/agent_prompting.py",
+			"src/builtin_agents.py",
+			"src/agent_runtime.py",
+		],
+		decision:
+			"子 Agent 负责观察和验证，主 Agent 负责用户上下文、最终决策和产物写入。",
+	},
 ];
 
 const skillSections: Section[] = [
@@ -311,6 +332,7 @@ export default function DocPage() {
 					<div>
 						<a href="#base">基座</a>
 						<a href="#loop">Agent Loop</a>
+						<a href="#subagents">子 Agent</a>
 						<a href="#skills">Skill</a>
 						<a href="#product">业务链路</a>
 						<a href="#examples">示例 Session</a>

frontend/app/public/doc-assets/technical-docs/11-subagents.md

@@ -0,0 +1,333 @@
+# 11. 子 Agent 使用边界和多 Agent 协作
+
+本文用于推进 ZK Data Agent 后续的子 Agent 能力增强。目标不是“让模型更爱开子 Agent”，而是把子 Agent 收敛为一种可靠的编排手段：只在长耗时、可并行、边界清晰、上下文可隔离的任务里使用。
+
+调研日期：2026-06-12。
+
+## 1. 外部实践共识
+
+### 1.1 Anthropic Research：适合宽度优先和并行探索
+
+Anthropic 的 Research 多 Agent 系统采用 lead agent + parallel subagents。Lead agent 负责分析用户问题、制定研究策略，再派出多个 subagent 同时探索不同方向。
+
+它强调几个判断点：
+
+- 多 Agent 特别适合开放式研究，因为研究路径不可提前硬编码，需要根据中间发现动态调整。
+- Subagent 可以拥有独立上下文窗口，并行探索不同方面，再把压缩后的结果交回主 Agent。
+- 多 Agent 在 breadth-first query 上更有效，尤其是多个方向可以同时查的时候。
+- 成本明显更高。Anthropic 提到 agent 通常比普通聊天消耗更多 token，多 Agent 系统消耗更高，因此只适合任务价值足够高的场景。
+- 如果所有 Agent 都必须共享同一上下文，或者子任务之间依赖很多，就不适合多 Agent。
+
+参考：https://www.anthropic.com/engineering/multi-agent-research-system
+
+### 1.2 OpenAI Agents SDK：handoff 是专业化分工
+
+OpenAI Agents SDK 的 handoff 用于把任务委托给另一个 Agent，特别适合不同 Agent 专注不同领域的场景。Handoff 本身以 tool 的形式暴露给 LLM，让模型决定是否转交。
+
+这给我们的启发是：
+
+- 子 Agent 不应该只是“再开一个一样的模型”，而应该有明确 profile。
+- Profile 描述要告诉模型什么时候使用该 Agent。
+- Handoff 可以带输入过滤和动态启用条件，避免把全量上下文无脑传给子 Agent。
+
+参考：https://openai.github.io/openai-agents-python/handoffs/
+
+### 1.3 LangChain / LangGraph：上下文工程比拆 Agent 更重要
+
+LangChain 对多 Agent 的总结非常贴近我们的情况：多 Agent 成败的关键不是数量，而是 context engineering。每个 subagent 需要明确目标、输出格式、工具范围、数据来源和边界，否则容易重复工作、遗漏信息或互相冲突。
+
+另一个重要观点是：以“读”为主的多 Agent 更容易成功，以“写”为主的多 Agent 更难。多个 Agent 同时写代码、写文档或写同一份数据，容易产生冲突；更好的方式是让多个子 Agent 读、查、分析，最后由主 Agent 统一合成和写入。
+
+LangGraph Supervisor 也采用 supervisor-worker 模式：单一 supervisor 负责用户交互，worker agent 只和 supervisor 通信。
+
+参考：
+
+- https://www.langchain.com/blog/how-and-when-to-build-multi-agent-systems
+- https://changelog.langchain.com/announcements/langgraph-supervisor-a-library-for-hierarchical-multi-agent-systems
+
+### 1.4 AutoGen：多 Agent 是编排框架，不是默认答案
+
+AutoGen 强调可对话、可定制、可集成工具和人工反馈的 Agent 协作。它适合构建复杂 LLM workflow，但也意味着需要显式设计对话模式、工具权限、人类检查点和终止条件。
+
+对 ZK Data Agent 来说，这说明多 Agent 能力应该服务于“工作流编排”，不能变成模型遇到复杂任务就随意分身。
+
+参考：https://microsoft.github.io/autogen/0.2/docs/Use-Cases/agent_chat/
+
+## 2. ZK Data Agent 当前机制
+
+当前子 Agent 的创建不是后端自动发生的，而是主模型调用 `Agent` 或旧名 `delegate_agent` 工具后触发。
+
+关键链路：
+
+```text
+主 Agent 进入 turn loop
+  -> 模型返回 tool_calls
+  -> tool_call.name == Agent / delegate_agent
+  -> _execute_delegate_agent(arguments)
+  -> 解析 subagent_type / prompt / subtasks / max_turns / allow_write
+  -> 创建 LocalCodingAgent 子实例
+  -> 子 Agent 完成后把 child_results / delegate_batches 写回工具结果
+  -> 主 Agent 继续总结或执行下一步
+```
+
+当前实现里的重要约束：
+
+- `subtasks` 最多取前 8 个。
+- 子 Agent 默认不能递归调用 `Agent` / `delegate_agent`。
+- Explore、Plan、verification 等只读 Agent 禁止写文件。
+- 写权限需要父 Agent 允许，并且工具参数显式 `allow_write=true`。
+- 子会话会标记 `session_metadata.visibility = child`，默认不进入左侧主会话列表。
+- 右侧活动区通过 `child_results`、`delegate_batches`、`dependency_skips` 展示汇总。
+
+相关代码：
+
+```text
+src/agent_tools.py            Agent 工具 schema 和描述
+src/agent_prompting.py        子 Agent 系统提示词注入
+src/builtin_agents.py         内置 Agent profile 和 when_to_use
+src/agent_runtime.py          _execute_delegate_agent / _run_single_subtask
+frontend/.../activity-panel   子 Agent 活动汇总展示
+```
+
+## 3. 我们的设计定位
+
+ZK Data Agent 里应该有三层能力：
+
+```text
+主 Agent
+  负责用户意图、当前会话状态、Skill 选择、最终决策、最终写入和回复。
+
+Skill
+  负责领域流程、业务知识、稳定脚本、review 门禁和产物规范。
+
+子 Agent
+  负责边界清晰的独立子任务，尤其是读、查、分析、验证、候选召回。
+```
+
+子 Agent 不是 Skill 的替代品，也不是主 Agent 的替代品。
+
+- Skill 回答“这类业务流程应该怎么做”。
+- Tool 回答“这个确定性动作怎么执行”。
+- 子 Agent 回答“这个可隔离子任务能不能交给另一个 Agent 独立完成”。
+
+## 4. 什么时候应该优先使用子 Agent
+
+### 4.1 明显适合
+
+满足下面多个条件时，可以优先考虑子 Agent：
+
+| 判断项 | 说明 | 示例 |
+|---|---|---|
+| 长耗时 | 主 Agent 直接串行做会占用很多轮 | 查多个索引、读大量文件、批量分析 badcase |
+| 可并行 | 子任务之间没有强依赖 | 同时分析 5 个 rid、同时对比 3 个标签边界 |
+| 可隔离 | 子任务只需要一小段输入，不依赖完整上下文 | 只给某个文件路径、某个 request_id、某个候选集合 |
+| 读多写少 | 子任务主要是检索、阅读、总结、验证 | Explore / Plan / verification |
+| 输出可压缩 | 子 Agent 结果能用结构化摘要交回主 Agent | 输出发现、证据、风险、建议 |
+| 价值足够高 | 多消耗 token 和时间是值得的 | 线上事故排查、复杂数据策略、发布前验证 |
+
+典型场景：
+
+- 多个 request_id / session / badcase 可以独立排查。
+- 多个文件目录需要分别探索，再合并成架构判断。
+- 一个任务需要先从多个候选方向召回证据。
+- 非平凡代码修改后，需要 verification agent 独立跑测试和检查。
+- 上下文窗口快满，且有一个清晰阶段可以交给新上下文继续探索。
+
+### 4.2 可以使用，但需要谨慎
+
+这些场景可以用子 Agent，但必须限制边界：
+
+- 需要生成多个候选方案，但最后只能采用一个方案。
+- 需要多个 Agent 分别给 review 意见，但最终修改只能由主 Agent 做。
+- 需要长时间跑工具或搜索，但要避免子 Agent 自己继续扩散任务。
+- 需要子 Agent 写临时文件，必须限定在当前 session 的 scratchpad 或 output，并明确 `allow_write=true`。
+
+### 4.3 不应该使用
+
+这些场景默认不要开子 Agent：
+
+| 场景 | 原因 |
+|---|---|
+| 一次普通问答或一次工具调用 | 额外开 Agent 只会增加延迟和 token |
+| 强依赖连续上下文 | 子 Agent 容易拿不到主会话里的隐含状态 |
+| 子任务之间依赖链很长 | 调度成本高，失败恢复复杂 |
+| 多个 Agent 同时写同一份产物 | 容易冲突，合并困难 |
+| 需要和用户持续互动 | 子 Agent 的交互会割裂主会话体验 |
+| 主 Agent 自己还没想清任务边界 | 边界不清时拆出去只会放大混乱 |
+| 业务流程已有明确 Skill | 应先用 Skill，而不是直接用通用子 Agent |
+
+## 5. 子 Agent 调用前检查表
+
+主 Agent 在调用 `Agent` 前，应在内部完成这个检查：
+
+```text
+1. 这个任务能否直接用一个工具完成？
+   能 -> 不开子 Agent。
+
+2. 这个任务是否命中明确 Skill？
+   是 -> 先调用 Skill，除非 Skill 内明确建议拆子 Agent。
+
+3. 是否存在独立子任务？
+   没有 -> 不开子 Agent。
+
+4. 子任务是否可以用少量输入描述清楚？
+   不能 -> 不开子 Agent，先由主 Agent 整理上下文。
+
+5. 子任务结果是否能用结构化摘要交回？
+   不能 -> 不开子 Agent。
+
+6. 子任务是否主要是读、查、分析、验证？
+   是 -> 适合。
+   否 -> 需要明确写权限、路径和合并策略。
+
+7. 是否有并行收益或上下文隔离收益？
+   没有 -> 不开子 Agent。
+```
+
+## 6. 子 Agent Prompt 契约
+
+每次委托都应该给子 Agent 一个完整契约。不要只写“帮我看看这个”。
+
+推荐格式：
+
+```text
+任务目标：
+  你要完成什么，不要做什么。
+
+输入材料：
+  文件路径、request_id、query、候选数据、时间范围等。
+
+上下文边界：
+  只依赖这些输入；不要假设完整主会话上下文。
+
+允许动作：
+  只读 / 可运行命令 / 可写临时文件 / 可写 output。
+
+禁止动作：
+  不修改平台代码；不提交 git；不改非本 session 文件。
+
+输出格式：
+  用固定字段返回，例如 findings、evidence、risks、next_actions。
+
+失败处理：
+  找不到数据时说明尝试过的路径和下一步建议。
+```
+
+示例：
+
+```text
+请作为只读 Explore 子 Agent，分析 request_id=xxx 的线上日志。
+只允许读取 ELK 查询结果和当前 session 文件，不写文件。
+目标是判断是否能还原 query、prev_session、模型 prompt、模型输出。
+输出 JSON：
+{
+  "request_id": "...",
+  "found_fields": [],
+  "missing_fields": [],
+  "evidence": [],
+  "recommended_next_query": ""
+}
+```
+
+## 7. 和 Skill 的组合方式
+
+### 7.1 Skill 内可以建议子 Agent，但不应该滥用
+
+Skill 可以在流程里写：
+
+- 当候选样本超过一定数量，可以用 Explore 子 Agent 并行抽样。
+- 当多个 rid 独立排查，可以每个 rid 一个子 Agent。
+- 当生成数据前需要多标签边界 review，可以让多个只读子 Agent 分别分析边界。
+
+但 Skill 不应该写：
+
+- 每次执行都必须开子 Agent。
+- 所有生成都交给子 Agent。
+- 让多个子 Agent 同时写最终数据。
+
+### 7.2 主 Agent 负责最终合成
+
+推荐模式：
+
+```text
+子 Agent A：查证据
+子 Agent B：查另一批证据
+子 Agent C：做只读验证
+  -> 主 Agent 汇总
+  -> 主 Agent 决定下一步
+  -> 主 Agent 调脚本写最终产物
+```
+
+不推荐：
+
+```text
+子 Agent A 写一半数据
+子 Agent B 写另一半数据
+子 Agent C 修改格式
+  -> 主 Agent 被动猜测哪个文件是最终版
+```
+
+## 8. UI 和可观测性原则
+
+子 Agent 不应该像普通会话一样挤进左侧列表。用户关心的是：
+
+- 主任务有没有拆子任务。
+- 拆了几个。
+- 每个子任务是否完成、失败、跳过。
+- 每个子任务的关键发现是什么。
+- 必要时能 drill down 到子会话详情。
+
+因此 UI 默认应该：
+
+- 左侧会话列表只显示主会话。
+- 右侧活动区展示子 Agent 汇总、批次、子任务摘要。
+- 子 Agent 详情默认折叠。
+- 子 Agent 输出应该有结构化摘要，不展示完整长日志。
+
+## 9. 后续增强方向
+
+### P0：收敛提示词和工具描述
+
+- 在系统提示词里明确“子 Agent 不是默认策略”。
+- 把“适合/不适合”的判断表加入子 Agent 指导。
+- `Agent` 工具描述中补充：优先用于独立、可并行、长耗时、读多写少的子任务。
+- 对 product-data、online-mining-v2、label-master 的 Skill 文档增加是否建议子 Agent 的边界。
+
+### P1：结构化委托和活动展示
+
+- 为 `Agent` 参数增加更明确的 `output_contract` 或 `expected_fields`。
+- 活动区展示子任务输入、状态、摘要、失败原因。
+- 支持从活动区打开 child session，但默认隐藏。
+- 对运行中的子 Agent 做更稳定的 live event 汇总。
+
+### P2：策略控制和评估
+
+- 增加每轮最多子 Agent 数量、并发数、总 token 预算。
+- 增加“只读子 Agent”默认模式。
+- 记录每次子 Agent 的收益：是否减少主任务耗时、是否提升结果质量、是否造成失败。
+- 建立小型 eval：同一任务单 Agent vs 子 Agent，对比准确率、耗时、token、用户体验。
+
+### P3：业务化多 Agent
+
+- online-mining-v2：多个 rid / 多个索引方向并行探索。
+- label-master：多个候选标签边界并行查证，然后主 Agent 汇总裁决。
+- product-data：批量数据质量 review 可并行，但最终 records 写入必须由主 Agent 或确定性脚本完成。
+
+## 10. 当前建议
+
+短期不要把多 Agent 做成自动默认能力。更稳的路线是：
+
+```text
+默认：主 Agent + Skill + Tool
+遇到明确并行探索 / 独立验证 / 长耗时检索
+  -> 主 Agent 显式调用子 Agent
+  -> 子 Agent 只返回结构化摘要
+  -> 主 Agent 合成、写入、回复
+```
+
+一句话原则：
+
+```text
+子 Agent 负责把复杂任务拆成可独立完成的观察和验证；
+主 Agent 负责保持用户上下文、做最终决策和产物写入。
+```

frontend/app/public/doc-assets/technical-docs/README.md

@@ -15,6 +15,7 @@
 8. [label-master Skill 实现](08-label-master.md)
 9. [外部系统 Skill：ELK、SQL、模型迭代](09-external-skills.md)
 10. [Agent 记忆机制调研与对比](10-memory-research.md)
+11. [子 Agent 使用边界和多 Agent 协作](11-subagents.md)
 
 ## 一句话定位
 

skills/online-replay-diff/SKILL.md

@@ -0,0 +1,107 @@
+---
+name: online-replay-diff
+description: 处理线上流量回放产生的 diff Excel 文件；当用户要处理线上系统回放 diff、计算 car/phone/glass 等回放数据 diff 率、整理 is_same=False 待标注样本，或复用 data-diff 的 fix_excel / diff_all / select_false 流程时使用。
+metadata:
+  short-description: 计算线上回放 diff 率并输出待标注 False 样本
+---
+
+使用这个 skill 处理线上流量回放导出的 Excel diff 文件：把 `.xls/.xlsx` 清洗成稳定表头，计算 `is_same`，统计 diff 率，并输出待人工标注的数据。
+
+## 输入形态
+
+常见输入是线上回放导出的老式 `.xls`：
+
+- 第 1 行是标题，例如 `DiffCase数据`。
+- 第 2 行是真实表头。
+- 必须包含 `Query`、`Preview环境domain`、`Ptr环境domain`。
+- 常见文件名：`car_random.xls`、`car_top.xls`、`phone_random.xls`、`glass_top.xls`。
+
+脚本也兼容已经清洗过的 `.xlsx`，会在前 10 行自动寻找真实表头。
+
+## 处理口径
+
+- `is_same=True`：`Preview环境domain` 与 `Ptr环境domain` 完全相同，或落在同一等价组。
+- `is_same=False`：两边 domain 不同且不在等价组内，进入待标注输出。
+- diff 率：`is_same=False 行数 / 总行数`。
+
+默认等价组在 `knowledge/equivalence_groups.json`：
+
+- `CalendarQA` / `time` / `TimeDistance`
+- `Chat` / `QA` / `dialogCopilot`
+- `controlCopilot` / `soundboxControl` / `smartMiot` / `smartApp:defaultApp` / `smartApp:app-commander` / `camera`
+- `music` / `station`
+
+如果用户给出新的等价关系，优先更新或另存一个 JSON，通过 `--equiv-config` 传入；不要直接在脚本里临时写死。
+
+## 推荐流程
+
+1. 确认输入文件和输出目录。用户没指定时，优先在当前工作目录下找 `.xls/.xlsx`，输出到 `output/`。
+2. 确认依赖可用：
+   ```bash
+   python - <<'PY'
+   import pandas, openpyxl, xlrd
+   print("excel deps ok")
+   PY
+   ```
+   缺依赖时，在项目虚拟环境里安装：`pip install pandas openpyxl xlrd==2.0.1`。
+3. 运行脚本：
+   ```bash
+   python /Users/wuyang/Project/claw-code-agent/skills/online-replay-diff/scripts/process_replay_diff.py \
+     car_random.xls car_top.xls \
+     --output-dir output \
+     --prefix car
+   ```
+4. 检查控制台 JSON 或 `*_summary.md`，向用户汇报每个文件和总体 diff 率。
+5. 把 `*_待标注_仅False.xlsx` 作为待标注数据交付。
+
+## 脚本
+
+```text
+skills/online-replay-diff/
+  SKILL.md
+  knowledge/
+    equivalence_groups.json
+  scripts/
+    process_replay_diff.py
+```
+
+### process_replay_diff.py
+
+参数：
+
+- 位置参数：输入 `.xls/.xlsx` 文件或目录。
+- `--input-dir`：输入目录；和位置参数目录等价。
+- `--glob`：目录扫描通配符，默认 `*.xls*`。
+- `--output-dir`：输出目录，默认 `output`。
+- `--prefix`：输出文件名前缀，默认 `replay_diff`。
+- `--equiv-config`：domain 等价规则 JSON。
+
+输出：
+
+- `normalized/<stem>.xlsx`：每个输入文件清洗首行标题后的单表。
+- `<prefix>_汇总结果.xlsx`：每个输入文件一个 sheet，补齐固定列并追加 `is_same`。
+- `<prefix>_待标注_仅False.xlsx`：只保留 `is_same=False` 的待标注数据。
+- `<prefix>_summary.json`：机器可读统计。
+- `<prefix>_summary.csv`：每文件统计表。
+- `<prefix>_summary.md`：可粘贴到在线文档的处理摘要。
+
+## 汇报格式
+
+处理完优先这样回复：
+
+```text
+已处理 N 个回放 diff 文件。
+
+- car_random：总行数 7,285，diff 416，diff 率 5.7104%
+- car_top：总行数 12,260，diff 485，diff 率 3.9560%
+- 总体：总行数 19,545，diff 901，diff 率 4.6099%
+
+待标注数据：/abs/path/output/car_待标注_仅False.xlsx
+汇总结果：/abs/path/output/car_汇总结果.xlsx
+```
+
+如果脚本失败，先检查：
+
+- 是否缺 `xlrd`，老式 `.xls` 必须用它读。
+- 是否找不到真实表头；前 10 行必须能看到 `Query`、`Preview环境domain`、`Ptr环境domain`。
+- 是否输入了脚本刚产出的汇总文件，导致重复处理输出文件。

skills/online-replay-diff/knowledge/equivalence_groups.json

@@ -0,0 +1,9 @@
+{
+  "groups": [
+    ["CalendarQA", "time", "TimeDistance"],
+    ["Chat", "QA", "dialogCopilot"],
+    ["controlCopilot", "soundboxControl", "smartMiot", "smartApp:defaultApp", "smartApp:app-commander", "camera"],
+    ["music", "station"]
+  ],
+  "pairs": []
+}

skills/online-replay-diff/scripts/process_replay_diff.py

@@ -0,0 +1,363 @@
+#!/usr/bin/env python3
+"""Process online traffic replay diff Excel files and export labeling workbooks."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import math
+import re
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+import pandas as pd
+
+
+SCRIPT_DIR = Path(__file__).resolve().parent
+SKILL_DIR = SCRIPT_DIR.parent
+DEFAULT_EQUIV_CONFIG = SKILL_DIR / "knowledge" / "equivalence_groups.json"
+
+REQUIRED_DOMAIN_COLUMNS = ["Preview环境domain", "Ptr环境domain"]
+
+KEEP_COLUMNS = [
+    "Query",
+    "PV",
+    "diff结果",
+    "reviewer",
+    "GSB",
+    "Preview环境requestId",
+    "Ptr环境requestId",
+    "Preview环境toSpeak",
+    "Ptr环境toSpeak",
+    "Preview环境domain",
+    "Ptr环境domain",
+    "Preview环境normCode",
+    "Ptr环境normCode",
+    "Preview环境funcCategory",
+    "Ptr环境funcCategory",
+    "Preview环境funcName",
+    "Ptr环境funcName",
+    "Preview环境agentType",
+    "Ptr环境agentType",
+    "Preview环境copilotCode",
+    "Ptr环境copilotCode",
+    "Preview环境promptString",
+    "Ptr环境promptString",
+    "Preview环境promptModel",
+    "Ptr环境promptModel",
+    "baseRequestId",
+    "回放时间",
+    "is_same",
+]
+
+
+@dataclass
+class FileResult:
+    source: Path
+    sheet_name: str
+    total: int
+    same: int
+    diff: int
+    diff_rate: float
+    normalized_path: Path
+
+
+def load_equivalence_config(path: Path) -> tuple[list[list[str]], list[tuple[str, str]]]:
+    if not path.exists():
+        raise FileNotFoundError(f"等价规则不存在: {path}")
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    groups = payload.get("groups", [])
+    pairs = payload.get("pairs", [])
+    if not isinstance(groups, list) or not isinstance(pairs, list):
+        raise ValueError("等价规则 JSON 必须包含 list 类型的 groups / pairs")
+    normalized_groups = [[str(item).strip() for item in group if str(item).strip()] for group in groups]
+    normalized_pairs = []
+    for pair in pairs:
+        if not isinstance(pair, list) or len(pair) != 2:
+            raise ValueError(f"pairs 中每项必须是长度为 2 的数组: {pair}")
+        normalized_pairs.append((str(pair[0]).strip(), str(pair[1]).strip()))
+    return normalized_groups, normalized_pairs
+
+
+def build_is_same(groups: list[list[str]], pairs: list[tuple[str, str]]):
+    parent: dict[str, str] = {}
+
+    def find(x: str) -> str:
+        parent.setdefault(x, x)
+        while parent[x] != x:
+            parent[x] = parent[parent[x]]
+            x = parent[x]
+        return x
+
+    def union(a: str, b: str) -> None:
+        ra, rb = find(a), find(b)
+        if ra != rb:
+            parent[rb] = ra
+
+    for group in groups:
+        if len(group) < 2:
+            continue
+        for item in group[1:]:
+            union(group[0], item)
+    for left, right in pairs:
+        union(left, right)
+
+    def normalize(value: Any) -> str | None:
+        if value is None:
+            return None
+        if isinstance(value, float) and math.isnan(value):
+            return None
+        text = str(value).strip()
+        return text if text else None
+
+    def is_same(left: Any, right: Any) -> bool:
+        left_text = normalize(left)
+        right_text = normalize(right)
+        if left_text is None or right_text is None:
+            return False
+        if left_text == right_text:
+            return True
+        return find(left_text) == find(right_text)
+
+    return is_same
+
+
+def collect_input_files(args: argparse.Namespace) -> list[Path]:
+    files: list[Path] = []
+    for item in args.inputs or []:
+        path = Path(item).expanduser().resolve()
+        if path.is_dir():
+            files.extend(sorted(path.glob(args.glob)))
+        else:
+            files.append(path)
+    if args.input_dir:
+        files.extend(sorted(Path(args.input_dir).expanduser().resolve().glob(args.glob)))
+    unique: list[Path] = []
+    seen = set()
+    for path in files:
+        if path.name.startswith("~$"):
+            continue
+        if path.suffix.lower() not in {".xls", ".xlsx"}:
+            continue
+        if path not in seen:
+            unique.append(path)
+            seen.add(path)
+    return unique
+
+
+def find_header_row(raw: pd.DataFrame) -> int:
+    max_scan = min(10, len(raw))
+    required = set(["Query", *REQUIRED_DOMAIN_COLUMNS])
+    for idx in range(max_scan):
+        row_values = {str(value).strip() for value in raw.iloc[idx].tolist() if not pd.isna(value)}
+        if required.issubset(row_values):
+            return idx
+    raise ValueError("前 10 行未找到包含 Query / Preview环境domain / Ptr环境domain 的表头")
+
+
+def make_unique_columns(columns: list[Any]) -> list[str]:
+    seen: dict[str, int] = {}
+    result = []
+    for col in columns:
+        name = str(col).strip()
+        if not name or name.lower() == "nan":
+            name = "Unnamed"
+        count = seen.get(name, 0)
+        seen[name] = count + 1
+        result.append(name if count == 0 else f"{name}.{count}")
+    return result
+
+
+def read_replay_excel(path: Path) -> pd.DataFrame:
+    engine = "xlrd" if path.suffix.lower() == ".xls" else "openpyxl"
+    raw = pd.read_excel(path, header=None, engine=engine)
+    header_row = find_header_row(raw)
+    df = raw.iloc[header_row + 1 :].copy()
+    df.columns = make_unique_columns(raw.iloc[header_row].tolist())
+    df = df.dropna(how="all").reset_index(drop=True)
+    missing = [col for col in REQUIRED_DOMAIN_COLUMNS if col not in df.columns]
+    if missing:
+        raise ValueError(f"{path.name} 缺少必要列: {missing}")
+    return df
+
+
+def sanitize_sheet_name(name: str) -> str:
+    name = re.sub(r"[\[\]:*?/\\]", "_", name).strip()
+    return name[:31] if name else "Sheet"
+
+
+def make_unique_sheet_name(name: str, used: set[str]) -> str:
+    base = sanitize_sheet_name(name)
+    candidate = base
+    index = 1
+    while candidate in used:
+        suffix = f"_{index}"
+        candidate = base[: 31 - len(suffix)] + suffix
+        index += 1
+    used.add(candidate)
+    return candidate
+
+
+def is_false_value(value: Any) -> bool:
+    if pd.isna(value):
+        return False
+    if isinstance(value, bool):
+        return value is False
+    if isinstance(value, (int, float)):
+        return value == 0
+    return str(value).strip().lower() in {"false", "0"}
+
+
+def write_markdown_summary(path: Path, results: list[FileResult]) -> None:
+    total = sum(item.total for item in results)
+    diff = sum(item.diff for item in results)
+    same = sum(item.same for item in results)
+    overall_rate = diff / total if total else 0.0
+    lines = [
+        "# 线上流量回放 diff 处理结果",
+        "",
+        "## 处理口径",
+        "",
+        "- 自动识别回放 Excel 中的真实表头，兼容首行 `DiffCase数据` 标题行。",
+        "- 按 `Preview环境domain` 与 `Ptr环境domain` 计算 `is_same`。",
+        "- 最终 diff 按 `is_same=False` 统计，待标注数据也只保留这些行。",
+        "",
+        "## diff 率",
+        "",
+        "| 文件 | 总行数 | same 数 | diff 数 | diff 率 |",
+        "|---|---:|---:|---:|---:|",
+    ]
+    for item in results:
+        lines.append(
+            f"| {item.sheet_name} | {item.total:,} | {item.same:,} | {item.diff:,} | {item.diff_rate:.4%} |"
+        )
+    lines.append(f"| 合计 | {total:,} | {same:,} | {diff:,} | {overall_rate:.4%} |")
+    lines.append("")
+    path.write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+
+def process_files(args: argparse.Namespace) -> dict[str, Any]:
+    inputs = collect_input_files(args)
+    if not inputs:
+        raise FileNotFoundError("未找到输入 Excel 文件")
+
+    output_dir = Path(args.output_dir).expanduser().resolve()
+    normalized_dir = output_dir / "normalized"
+    output_dir.mkdir(parents=True, exist_ok=True)
+    normalized_dir.mkdir(parents=True, exist_ok=True)
+
+    groups, pairs = load_equivalence_config(Path(args.equiv_config).expanduser().resolve())
+    is_same = build_is_same(groups, pairs)
+
+    used_sheet_names: set[str] = set()
+    processed: list[tuple[str, pd.DataFrame]] = []
+    results: list[FileResult] = []
+
+    for input_path in inputs:
+        df = read_replay_excel(input_path)
+        normalized_path = normalized_dir / f"{input_path.stem}.xlsx"
+        df.to_excel(normalized_path, index=False, engine="openpyxl")
+
+        df["is_same"] = df.apply(
+            lambda row: is_same(row["Preview环境domain"], row["Ptr环境domain"]),
+            axis=1,
+        )
+        for col in KEEP_COLUMNS:
+            if col not in df.columns:
+                df[col] = ""
+        result_df = df[KEEP_COLUMNS].copy()
+        sheet_name = make_unique_sheet_name(input_path.stem, used_sheet_names)
+        total = len(result_df)
+        diff = int(result_df["is_same"].apply(is_false_value).sum())
+        same = int((result_df["is_same"] == True).sum())
+        rate = diff / total if total else 0.0
+        processed.append((sheet_name, result_df))
+        results.append(
+            FileResult(
+                source=input_path,
+                sheet_name=sheet_name,
+                total=total,
+                same=same,
+                diff=diff,
+                diff_rate=rate,
+                normalized_path=normalized_path,
+            )
+        )
+
+    summary_workbook = output_dir / f"{args.prefix}_汇总结果.xlsx"
+    false_workbook = output_dir / f"{args.prefix}_待标注_仅False.xlsx"
+    summary_json = output_dir / f"{args.prefix}_summary.json"
+    summary_csv = output_dir / f"{args.prefix}_summary.csv"
+    summary_md = output_dir / f"{args.prefix}_summary.md"
+
+    with pd.ExcelWriter(summary_workbook, engine="openpyxl") as writer:
+        for sheet_name, df in processed:
+            df.to_excel(writer, sheet_name=sheet_name, index=False)
+
+    with pd.ExcelWriter(false_workbook, engine="openpyxl") as writer:
+        kept = 0
+        for sheet_name, df in processed:
+            filtered_df = df[df["is_same"].apply(is_false_value)].copy()
+            if not filtered_df.empty:
+                filtered_df.to_excel(writer, sheet_name=sheet_name, index=False)
+                kept += 1
+        if kept == 0:
+            pd.DataFrame({"说明": ["所有 sheet 都没有 is_same=False 的数据"]}).to_excel(
+                writer,
+                sheet_name="result",
+                index=False,
+            )
+
+    summary_rows = [
+        {
+            "file": str(item.source),
+            "sheet": item.sheet_name,
+            "total": item.total,
+            "same": item.same,
+            "diff": item.diff,
+            "diff_rate": item.diff_rate,
+            "normalized_file": str(item.normalized_path),
+        }
+        for item in results
+    ]
+    total = sum(item.total for item in results)
+    diff = sum(item.diff for item in results)
+    same = sum(item.same for item in results)
+    payload = {
+        "total": total,
+        "same": same,
+        "diff": diff,
+        "diff_rate": diff / total if total else 0.0,
+        "files": summary_rows,
+        "outputs": {
+            "summary_workbook": str(summary_workbook),
+            "false_workbook": str(false_workbook),
+            "summary_json": str(summary_json),
+            "summary_csv": str(summary_csv),
+            "summary_md": str(summary_md),
+        },
+    }
+    summary_json.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
+    pd.DataFrame(summary_rows).to_csv(summary_csv, index=False, encoding="utf-8-sig")
+    write_markdown_summary(summary_md, results)
+    return payload
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="处理线上流量回放 diff Excel，计算 diff 率并输出待标注数据")
+    parser.add_argument("inputs", nargs="*", help="输入 .xls/.xlsx 文件或目录")
+    parser.add_argument("--input-dir", help="输入目录；会按 --glob 扫描")
+    parser.add_argument("--glob", default="*.xls*", help="目录扫描通配符，默认 *.xls*")
+    parser.add_argument("--output-dir", default="output", help="输出目录，默认 ./output")
+    parser.add_argument("--prefix", default="replay_diff", help="输出文件名前缀")
+    parser.add_argument("--equiv-config", default=str(DEFAULT_EQUIV_CONFIG), help="domain 等价规则 JSON")
+    args = parser.parse_args()
+
+    payload = process_files(args)
+    print(json.dumps(payload, ensure_ascii=False, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())