diff --git a/skills/online-mining-v2/README.md b/skills/online-mining-v2/README.md
new file mode 100644
index 0000000..92808fa
--- /dev/null
+++ b/skills/online-mining-v2/README.md
@@ -0,0 +1,38 @@
+# online-mining-v2
+
+`online-mining-v2` 是线上数据挖掘 skill。它把 `elk-fetch` 作为默认线上数据源，把 `product-data` 作为挖掘结果的标准数据后处理链路。
+
+## 设计目标
+
+- 线上数据源默认走 ELK，而不是本地 parquet。
+- 挖掘能力放在 skill 目录内，不注册新的平台工具。
+- 所有脚本支持 JSON stdin/stdout，方便迁移到其他 Agent。
+- 挖出的线上 case 不直接手写最终格式，而是转成 `product-data` 的 dataset draft，再进入 canonical record、校验和导出。
+
+## 主要脚本
+
+| 脚本 | 作用 |
+| --- | --- |
+| `scripts/elk_profile_index.py` | 采样索引 schema 和关键字段，帮助 Agent 判断字段可用性。 |
+| `scripts/elk_search_cases.py` | 按 query/domain/prompt/model output/candidate domain 等条件搜索候选 case。 |
+| `scripts/elk_fetch_by_request_ids.py` | 批量按 request id 拉取 main/pre_processing 原始摘要。 |
+| `scripts/elk_join_request_logs.py` | 把 `arch-flat-nlp-log-f-*` 和 `pre-processing-info*` 按 request id 合并。 |
+| `scripts/build_dataset_draft.py` | 把 review 后的线上 case 转成 product-data dataset draft text。 |
+
+## 数据源
+
+- `main`：`arch-flat-nlp-log-f-*`
+- `pre_processing`：`pre-processing-info*`
+
+字段说明见 `knowledge/数据源字段说明.md`。
+
+## 后处理
+
+后处理复用 `skills/product-data/scripts/`：
+
+1. `normalize_dataset_draft.py`
+2. `validate_dataset_records.py`
+3. `export_dataset_records.py`
+4. 可选 `export_training_jsonl.py`
+5. 可选 `export_planning_eval_csv.py`
+
diff --git a/skills/online-mining-v2/SKILL.md b/skills/online-mining-v2/SKILL.md
new file mode 100644
index 0000000..0d340f2
--- /dev/null
+++ b/skills/online-mining-v2/SKILL.md
@@ -0,0 +1,204 @@
+---
+name: online-mining-v2
+description: 使用 ELK 作为默认线上数据源挖掘线上 case，并复用 product-data 的 canonical record 与导出流程。
+when_to_use: 当用户希望从线上日志中按 query/domain/model prompt/model output/候选 domain 等条件挖掘样本、review 命中质量、形成专项评测集或训练数据时使用。
+aliases: online-mining-elk, elk-mining, 线上挖掘v2
+allowed_tools: read_file, write_file, grep_search, glob_search, ask_user_question, python_exec, python_package
+---
+
+# Online Mining v2
+
+使用这个 skill 处理“线上日志需求 -> ELK 挖掘策略 -> 候选样本 review -> product-data 标准数据”的工作流。
+
+本 skill 不新增平台注册工具。能力都在 `skills/online-mining-v2/scripts/` 下，通过 `python_exec` 的 `script_path` 模式执行；如果迁移到其他 Agent，也可以直接运行这些脚本。
+
+## 能力组织
+
+```text
+skills/online-mining-v2/
+  SKILL.md
+  README.md
+  knowledge/
+    数据源字段说明.md
+    工作流.md
+  scripts/
+    online_mining_common.py
+    elk_profile_index.py
+    elk_search_cases.py
+    elk_fetch_by_request_ids.py
+    elk_join_request_logs.py
+    build_dataset_draft.py
+```
+
+默认数据源：
+
+- `pre-processing-info*`：前处理日志，适合拿模型 prompt、模型输出、候选 domain、`excellent_domains_result`。
+- `arch-flat-nlp-log-f-*`：主 NLP 日志，适合拿最终 `query`、`domain`、`func`、`session_id`、`device_id`、`device`、`tts/text/to_speak`。
+
+后处理全部复用 `product-data`：
+
+- `skills/product-data/scripts/normalize_dataset_draft.py`
+- `skills/product-data/scripts/validate_dataset_records.py`
+- `skills/product-data/scripts/export_dataset_records.py`
+- `skills/product-data/scripts/export_training_jsonl.py`
+- `skills/product-data/scripts/export_planning_eval_csv.py`
+
+## 依赖
+
+如果 `python_exec` 返回缺少 `elasticsearch` 或 `urllib3`，先安装：
+
+```json
+{
+  "action": "install",
+  "packages": ["elasticsearch<8", "urllib3"],
+  "timeout_seconds": 120
+}
+```
+
+不要用 `bash` 手写临时脚本查询 ELK；优先使用本 skill 的 portable scripts。
+
+## 关键工作流
+
+### 1. 先理解需求，不直接挖
+
+从用户输入整理挖掘目标：
+
+- 目标问题：要评估准确率、构建评测集，还是补充训练数据。
+- 目标标签：如 `Agent(tag="地图导航")`、`Summarize()`。
+- 复杂度：`complex=true/false` 是否明确。
+- 线上筛选特征：query 关键词/正则、domain、func、候选 domain、模型输出、prompt 特征、设备、日期。
+- review 策略：先抽多少条给用户看，通常先 20 到 100 条。
+
+缺少目标标签或关键筛选字段时，先问用户，不要直接导出数据。
+
+### 2. 探索表和字段
+
+需要确认表字段时，执行：
+
+```json
+{
+  "script_path": "skills/online-mining-v2/scripts/elk_profile_index.py",
+  "stdin": {
+    "sources": ["main", "pre_processing"],
+    "date": "20260512",
+    "sample_size": 20
+  },
+  "timeout_seconds": 120,
+  "max_output_chars": 20000
+}
+```
+
+### 3. 搜索候选
+
+按策略搜索：
+
+```json
+{
+  "script_path": "skills/online-mining-v2/scripts/elk_search_cases.py",
+  "stdin": {
+    "source": "pre_processing",
+    "date": "20260512",
+    "size": 50,
+    "scan_size": 500,
+    "filters": {
+      "query_contains": ["总结一下"],
+      "planning_result_contains": ["Summarize"]
+    }
+  },
+  "timeout_seconds": 180,
+  "max_output_chars": 30000
+}
+```
+
+`source=main` 适合按最终 `domain` / `func` / `query` 搜索。`source=pre_processing` 适合按 `planningOriginalResult`、`promptModel`、`excellent_domains_result`、候选 domain 搜索。
+
+### 4. 补全双表信息
+
+候选里只有一张表信息时，用 request id 补全：
+
+```json
+{
+  "script_path": "skills/online-mining-v2/scripts/elk_join_request_logs.py",
+  "stdin": {
+    "request_ids": ["xxx", "yyy"],
+    "date": "20260512"
+  },
+  "timeout_seconds": 180,
+  "max_output_chars": 30000
+}
+```
+
+### 5. 给用户 review
+
+展示候选样本时只展示必要字段：
+
+- request_id
+- query
+- main.domain / main.func
+- pre.planning_result
+- pre.hit_rules
+- pre.candidate_domains
+- session_id / device_id
+- tts/text
+
+不要一次贴大量 prompt。需要看 prompt 时只展示截断摘要，或保存到会话 output 目录。
+
+### 6. 用户确认后转成 product-data draft
+
+线上候选直接作为样本时，先生成 dataset draft text：
+
+```json
+{
+  "script_path": "skills/online-mining-v2/scripts/build_dataset_draft.py",
+  "stdin": {
+    "dataset_label": "总结类线上专项",
+    "target": "Summarize()",
+    "complex": false,
+    "cases": [/* elk_search_cases 或 join 输出里的 cases */],
+    "output_path": "scratchpad/mined_dataset_draft.txt"
+  },
+  "timeout_seconds": 120,
+  "max_output_chars": 20000
+}
+```
+
+然后必须调用 product-data：
+
+```json
+{
+  "script_path": "skills/product-data/scripts/normalize_dataset_draft.py",
+  "stdin": {
+    "draft_path": "scratchpad/mined_dataset_draft.txt",
+    "source_type": "online",
+    "records_output_path": "scratchpad/normalized_records.jsonl",
+    "append": false,
+    "return_records": false
+  },
+  "timeout_seconds": 120,
+  "max_output_chars": 20000
+}
+```
+
+再执行校验和导出：
+
+- `validate_dataset_records.py`
+- `export_dataset_records.py`，输出固定 `output/records.jsonl`
+- 用户需要训练或评测时，再导出 `training.jsonl` 或 `eval_planning.csv`
+
+## 分支约束
+
+- 用户要“线上数据直接做样本/评测集”：只做挖掘、review、转换，不生成新 query。
+- 用户明确要“基于线上问题补数据/扩写/构造”：先完成线上问题 review，再切到 `product-data` 的 generation plan 流程。
+- 不要把挖掘策略第一次命中的样本直接当最终数据；至少展示一小批给用户 review。
+
+## 输出目录约束
+
+所有产物写当前会话目录：
+
+- 中间候选：`scratchpad/mined_candidates.jsonl`
+- joined 详情：`scratchpad/joined_cases.jsonl`
+- draft：`scratchpad/mined_dataset_draft.txt`
+- canonical records：通过 product-data 导出到 `output/records.jsonl`
+
+不要写项目根目录、`tasks/`、`src/` 或 `skills/`。
+
diff --git a/skills/online-mining-v2/knowledge/工作流.md b/skills/online-mining-v2/knowledge/工作流.md
new file mode 100644
index 0000000..b1622c1
--- /dev/null
+++ b/skills/online-mining-v2/knowledge/工作流.md
@@ -0,0 +1,54 @@
+# 工作流
+
+## 分支 A：线上候选直接作为数据样本
+
+适用表达：
+
+- “把筛选出的数据变成样本”
+- “拿线上数据做评测集”
+- “保留这批线上 case”
+- “导出候选样本”
+
+流程：
+
+1. 分析需求，确认目标标签和复杂度。
+2. 设计 ELK 挖掘策略。
+3. 用 `elk_search_cases.py` 拉候选。
+4. 用 `elk_join_request_logs.py` 补全 main/pre_processing 字段。
+5. 展示小批候选给用户 review。
+6. 用户确认后，用 `build_dataset_draft.py` 转 dataset draft。
+7. 用 `product-data` 脚本 normalize、validate、export。
+
+这个分支不生成新 query。
+
+## 分支 B：基于线上问题补充生成数据
+
+适用表达：
+
+- “基于这些 badcase 再生成一批”
+- “扩写类似 case”
+- “构造训练数据”
+
+流程：
+
+1. 先完成分支 A 的线上问题定位和 review。
+2. 总结线上错误模式、边界和负例。
+3. 切到 `product-data` 的 generation goal / generation plan。
+4. 用户确认计划后再生成新数据。
+
+## review 展示建议
+
+每条候选优先展示：
+
+```text
+request_id:
+query:
+planning_result:
+main.domain / main.func:
+candidate_domains:
+session_id / device_id:
+tts:
+```
+
+不要默认展示完整 prompt。prompt 很长时只展示命中片段。
+
diff --git a/skills/online-mining-v2/knowledge/数据源字段说明.md b/skills/online-mining-v2/knowledge/数据源字段说明.md
new file mode 100644
index 0000000..701f3fb
--- /dev/null
+++ b/skills/online-mining-v2/knowledge/数据源字段说明.md
@@ -0,0 +1,74 @@
+# 数据源字段说明
+
+## main：`arch-flat-nlp-log-f-*`
+
+适合拿最终线上请求结果。
+
+常用字段：
+
+| 字段 | 含义 |
+| --- | --- |
+| `request_id` | 请求 id，和 product-data 的 `request_id` 对齐。 |
+| `timestamp` | 请求时间戳。 |
+| `query` | 当前 query。 |
+| `domain` | 最终 domain。 |
+| `func` / `func_name` | 最终 function 或函数名。 |
+| `session_id` | 会话 id，可用于后续重建多轮。 |
+| `device_id` | 设备 id。 |
+| `device` | 设备 JSON，包含设备类型、经纬度等。 |
+| `text` / `display_text` / `to_speak` | 小爱回复文本，可作为上一轮 tts。 |
+| `intention` | 传统意图结果。 |
+| `intention.intent_arbitrator_info` | 中控仲裁信息，含 `llm_agent_info`、`score_domains`、`hit_rules` 等。 |
+| `llm_*` | 大模型分类相关字段，覆盖情况取决于链路。 |
+
+适合筛选：
+
+- `query` 包含/正则。
+- `domain`、`func`、`func_name`。
+- `session_id`、`device_id`、设备类型。
+- `intention.intent_arbitrator_info.llm_agent_info.agentType`。
+
+## pre_processing：`pre-processing-info*`
+
+适合拿前处理和规划模型判断过程。
+
+常用字段：
+
+| 路径 | 含义 |
+| --- | --- |
+| `requestId` | 请求 id。 |
+| `timestamp` | 前处理日志时间戳。 |
+| `responseBoby.nodes.0.core.query.query` | 当前 query。注意字段名是 `responseBoby`。 |
+| `responseBoby.nodes.0.core.dispatch_large_model_result.code` | 前处理最终 code。 |
+| `responseBoby.nodes.0.core.dispatch_large_model_result.dispatchLargeModelInput.promptModel` | planning 模型 prompt。 |
+| `responseBoby.nodes.0.core.dispatch_large_model_result.dispatchLargeModelInput.hitRuleList` | 触发规则。 |
+| `responseBoby.nodes.0.core.dispatch_large_model_result.planningDebugInfo.planningOriginalResult` | planning 模型原始输出。 |
+| `responseBoby.nodes.0.core.excellent_domains_result` | 候选 domain 及旧意图结果。 |
+| `responseBoby.nodes.0.properties.0.debug.domain_infos` | 前处理召回 domain 列表。 |
+| `responseBoby.nodes.0.properties.0.debug.promptString` | agent LLM prompt。 |
+| `responseBoby.nodes.0.properties.0.debug.planningPrompt` | planning prompt。 |
+| `responseBoby.nodes.0.properties.0.agentLLMIntent` | agent LLM 判定。 |
+
+适合筛选：
+
+- `query` 包含/正则。
+- `planningOriginalResult` 包含某标签或 function。
+- `promptModel` 包含某上下文、历史或 query 结构。
+- `excellent_domains_result[*].domain` 包含某候选 domain。
+- `domain_infos[*].domain` 包含某召回 domain。
+
+## join 规则
+
+两张表通过：
+
+```text
+main.request_id == pre_processing.requestId
+```
+
+join 后优先使用：
+
+- query：优先 `pre_processing` 的 query，缺失时用 `main.query`。
+- timestamp：优先 `main.timestamp`，缺失时用 `pre_processing.timestamp`。
+- tts：优先 `main.to_speak`，再 `main.text` 或 `display_text`。
+- label 参考：模型输出用 `pre_processing.planning_result`，线上最终结果用 `main.domain` / `main.func`。
+
diff --git a/skills/online-mining-v2/scripts/build_dataset_draft.py b/skills/online-mining-v2/scripts/build_dataset_draft.py
new file mode 100644
index 0000000..adc7758
--- /dev/null
+++ b/skills/online-mining-v2/scripts/build_dataset_draft.py
@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+import json
+
+from online_mining_common import (
+    emit_error,
+    emit_success,
+    load_json_payload,
+    resolve_portable_path,
+)
+
+
+def load_cases(payload: dict) -> list[dict]:
+    if isinstance(payload.get("cases"), list):
+        return [item for item in payload["cases"] if isinstance(item, dict)]
+    cases_path = payload.get("cases_path")
+    if not cases_path:
+        raise ValueError("cases or cases_path is required")
+    rows = []
+    text = resolve_portable_path(str(cases_path)).read_text(encoding="utf-8")
+    for line in text.splitlines():
+        if not line.strip():
+            continue
+        item = json.loads(line)
+        if isinstance(item, dict):
+            rows.append(item)
+    return rows
+
+
+def case_value(case: dict, key: str):
+    if key in case:
+        return case.get(key)
+    main = case.get("main") if isinstance(case.get("main"), dict) else {}
+    pre = case.get("pre_processing") if isinstance(case.get("pre_processing"), dict) else {}
+    return case.get(key) or pre.get(key) or main.get(key)
+
+
+def main() -> int:
+    try:
+        payload = load_json_payload()
+        dataset_label = str(payload.get("dataset_label") or "").strip()
+        target = str(payload.get("target") or "").strip()
+        if not dataset_label:
+            raise ValueError("dataset_label is required")
+        if not target:
+            raise ValueError("target is required")
+        complex_value = bool(payload.get("complex", False))
+        cases = load_cases(payload)
+        if not cases:
+            raise ValueError("cases must not be empty")
+
+        lines = [f"# dataset_label: {dataset_label}", ""]
+        for index, case in enumerate(cases, start=1):
+            query = str(case_value(case, "query") or "").strip()
+            if not query:
+                continue
+            request_id = str(case_value(case, "request_id") or "").strip()
+            timestamp = case_value(case, "timestamp")
+            tts = str(case_value(case, "tts") or "").strip()
+            planning_result = str(case_value(case, "planning_result") or "").strip()
+            notes = []
+            if planning_result:
+                notes.append(f"线上模型输出: {planning_result}")
+            main_domain = case_value(case, "domain") or case.get("main_domain")
+            if main_domain:
+                notes.append(f"线上 domain: {main_domain}")
+            lines.append(f"### case: online_{index:04d}")
+            lines.append(f"用户: {query}")
+            lines.append(f"complex: {'true' if complex_value else 'false'}")
+            lines.append(f"target: {target}")
+            if request_id:
+                lines.append(f"request_id: {request_id}")
+            if timestamp:
+                lines.append(f"timestamp: {timestamp}")
+            if tts:
+                lines.append(f"notes: tts={tts}" + (f"；{'；'.join(notes)}" if notes else ""))
+            elif notes:
+                lines.append(f"notes: {'；'.join(notes)}")
+            lines.append("")
+
+        draft_text = "\n".join(lines).strip() + "\n"
+        output_path = None
+        if payload.get("output_path"):
+            output = resolve_portable_path(str(payload["output_path"]))
+            output.parent.mkdir(parents=True, exist_ok=True)
+            output.write_text(draft_text, encoding="utf-8")
+            output_path = str(output)
+        emit_success({"case_count": draft_text.count("### case:"), "draft_text": draft_text, "output_path": output_path})
+        return 0
+    except Exception as exc:  # noqa: BLE001
+        emit_error(exc)
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
+
diff --git a/skills/online-mining-v2/scripts/elk_fetch_by_request_ids.py b/skills/online-mining-v2/scripts/elk_fetch_by_request_ids.py
new file mode 100644
index 0000000..a5f9a8d
--- /dev/null
+++ b/skills/online-mining-v2/scripts/elk_fetch_by_request_ids.py
@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+from online_mining_common import (
+    compact_text,
+    emit_error,
+    emit_success,
+    extract_case,
+    fetch_one_by_request_id,
+    load_json_payload,
+    string_values,
+    write_optional_jsonl,
+)
+
+
+def main() -> int:
+    try:
+        payload = load_json_payload()
+        request_ids = string_values(payload.get("request_ids"))
+        if not request_ids:
+            raise ValueError("request_ids is required")
+        sources = payload.get("sources") or ["main", "pre_processing"]
+        date = payload.get("date")
+        rows = []
+        for request_id in request_ids:
+            item = {"request_id": request_id}
+            for source in sources:
+                doc = fetch_one_by_request_id(str(source), request_id, date)
+                item[str(source)] = extract_case(str(source), doc) if doc else None
+            rows.append(item)
+        output_path = write_optional_jsonl(str(payload.get("output_path") or ""), rows)
+        emit_success(
+            {
+                "date": date or "past-48h",
+                "count": len(rows),
+                "output_path": output_path,
+                "cases": [
+                    {
+                        source: (
+                            {
+                                key: compact_text(value, 1000)
+                                for key, value in (case or {}).items()
+                                if key != "raw" and value not in (None, "", [], {})
+                            }
+                            if isinstance(case, dict)
+                            else case
+                        )
+                        for source, case in row.items()
+                    }
+                    for row in rows
+                ],
+            }
+        )
+        return 0
+    except Exception as exc:  # noqa: BLE001
+        emit_error(exc)
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
+
diff --git a/skills/online-mining-v2/scripts/elk_join_request_logs.py b/skills/online-mining-v2/scripts/elk_join_request_logs.py
new file mode 100644
index 0000000..0079d33
--- /dev/null
+++ b/skills/online-mining-v2/scripts/elk_join_request_logs.py
@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+import json
+
+from online_mining_common import (
+    compact_text,
+    emit_error,
+    emit_success,
+    extract_case,
+    fetch_one_by_request_id,
+    load_json_payload,
+    resolve_portable_path,
+    string_values,
+    write_optional_jsonl,
+)
+
+
+def load_request_ids(payload: dict) -> list[str]:
+    ids = string_values(payload.get("request_ids"))
+    if ids:
+        return ids
+    cases_path = payload.get("cases_path")
+    if cases_path:
+        request_ids = []
+        text = resolve_portable_path(str(cases_path)).read_text(encoding="utf-8")
+        for line in text.splitlines():
+            if not line.strip():
+                continue
+            item = json.loads(line)
+            rid = item.get("request_id") or item.get("requestId")
+            if rid:
+                request_ids.append(str(rid))
+        return request_ids
+    cases = payload.get("cases") or []
+    if isinstance(cases, list):
+        return [str(item.get("request_id")) for item in cases if isinstance(item, dict) and item.get("request_id")]
+    return []
+
+
+def merge_case(request_id: str, main_case: dict | None, pre_case: dict | None) -> dict:
+    return {
+        "request_id": request_id,
+        "query": (pre_case or {}).get("query") or (main_case or {}).get("query"),
+        "timestamp": (main_case or {}).get("timestamp") or (pre_case or {}).get("timestamp"),
+        "session_id": (main_case or {}).get("session_id"),
+        "device_id": (main_case or {}).get("device_id"),
+        "device": (main_case or {}).get("device"),
+        "tts": (main_case or {}).get("text"),
+        "main": main_case,
+        "pre_processing": pre_case,
+    }
+
+
+def main() -> int:
+    try:
+        payload = load_json_payload()
+        request_ids = load_request_ids(payload)
+        if not request_ids:
+            raise ValueError("request_ids, cases or cases_path is required")
+        date = payload.get("date")
+        rows = []
+        for request_id in request_ids:
+            main_doc = fetch_one_by_request_id("main", request_id, date)
+            pre_doc = fetch_one_by_request_id("pre_processing", request_id, date)
+            main_case = extract_case("main", main_doc) if main_doc else None
+            pre_case = extract_case("pre_processing", pre_doc) if pre_doc else None
+            rows.append(merge_case(request_id, main_case, pre_case))
+        output_path = write_optional_jsonl(str(payload.get("output_path") or ""), rows)
+        emit_success(
+            {
+                "date": date or "past-48h",
+                "count": len(rows),
+                "output_path": output_path,
+                "cases": [
+                    {
+                        key: compact_text(value, 1200)
+                        for key, value in row.items()
+                        if key not in {"main", "pre_processing"} and value not in (None, "", [], {})
+                    }
+                    | {
+                        "main_domain": (row.get("main") or {}).get("domain"),
+                        "main_func": (row.get("main") or {}).get("func"),
+                        "planning_result": (row.get("pre_processing") or {}).get("planning_result"),
+                        "candidate_domains": (row.get("pre_processing") or {}).get("candidate_domains"),
+                    }
+                    for row in rows
+                ],
+            }
+        )
+        return 0
+    except Exception as exc:  # noqa: BLE001
+        emit_error(exc)
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
+
diff --git a/skills/online-mining-v2/scripts/elk_profile_index.py b/skills/online-mining-v2/scripts/elk_profile_index.py
new file mode 100644
index 0000000..06698db
--- /dev/null
+++ b/skills/online-mining-v2/scripts/elk_profile_index.py
@@ -0,0 +1,75 @@
+from __future__ import annotations
+
+from online_mining_common import (
+    build_client,
+    compact_text,
+    deep_parse,
+    emit_error,
+    emit_success,
+    extract_case,
+    load_json_payload,
+    search_docs,
+    source_config,
+)
+
+
+def main() -> int:
+    try:
+        payload = load_json_payload()
+        sources = payload.get("sources") or ["main", "pre_processing"]
+        date = payload.get("date")
+        sample_size = int(payload.get("sample_size") or 5)
+        result: dict[str, object] = {}
+        for source in sources:
+            config = source_config(str(source))
+            client = build_client(str(source))
+            caps = client.field_caps(index=str(config["index"]), fields="*")
+            fields = caps.get("fields", {})
+            docs = search_docs(source=str(source), date=date, size=sample_size)
+            cases = [extract_case(str(source), deep_parse(doc)) for doc in docs]
+            interesting = [
+                name
+                for name in fields
+                if any(
+                    token in name.lower()
+                    for token in [
+                        "request",
+                        "query",
+                        "session",
+                        "domain",
+                        "func",
+                        "device",
+                        "prompt",
+                        "planning",
+                        "dispatch",
+                        "excellent",
+                        "llm",
+                        "timestamp",
+                    ]
+                )
+            ]
+            result[str(source)] = {
+                "index": config["index"],
+                "id_field": config["id_field"],
+                "time_field": config["time_field"],
+                "field_count": len(fields),
+                "interesting_fields": sorted(interesting)[:200],
+                "samples": [
+                    {
+                        key: compact_text(value, 600)
+                        for key, value in case.items()
+                        if key != "raw" and value not in (None, "", [], {})
+                    }
+                    for case in cases
+                ],
+            }
+        emit_success({"sources": result})
+        return 0
+    except Exception as exc:  # noqa: BLE001
+        emit_error(exc)
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
+
diff --git a/skills/online-mining-v2/scripts/elk_search_cases.py b/skills/online-mining-v2/scripts/elk_search_cases.py
new file mode 100644
index 0000000..a599833
--- /dev/null
+++ b/skills/online-mining-v2/scripts/elk_search_cases.py
@@ -0,0 +1,86 @@
+from __future__ import annotations
+
+from online_mining_common import (
+    case_matches,
+    compact_text,
+    emit_error,
+    emit_success,
+    extract_case,
+    load_json_payload,
+    search_docs,
+    source_config,
+    string_values,
+    write_optional_jsonl,
+)
+
+
+def build_index_filters(source: str, filters: dict) -> list[dict]:
+    """尽量把可索引条件下推到 ES，其余条件在客户端二次过滤。"""
+    config = source_config(source)
+    result: list[dict] = []
+    if source == "main":
+        for key in ("domain", "func"):
+            values = string_values(filters.get(key))
+            if values:
+                result.append({"terms": {key: values}})
+        contains = string_values(filters.get("query_contains"))
+        if contains:
+            # query 是 keyword 字段，wildcard 可用但不要放太宽，调用方要限制日期和 scan_size。
+            for item in contains:
+                result.append({"wildcard": {"query": {"value": f"*{item}*"}}})
+    if filters.get("request_id"):
+        result.append({"terms": {str(config["id_field"]): string_values(filters.get("request_id"))}})
+    return result
+
+
+def main() -> int:
+    try:
+        payload = load_json_payload()
+        source = str(payload.get("source") or "pre_processing")
+        date = payload.get("date")
+        size = int(payload.get("size") or 50)
+        scan_size = int(payload.get("scan_size") or max(size * 5, size))
+        filters = payload.get("filters") or {}
+        if not isinstance(filters, dict):
+            raise ValueError("filters must be an object")
+        docs = search_docs(
+            source=source,
+            date=date,
+            size=scan_size,
+            query_filters=build_index_filters(source, filters),
+        )
+        cases = []
+        for doc in docs:
+            case = extract_case(source, doc)
+            if not case_matches(case, filters):
+                continue
+            cases.append(case)
+            if len(cases) >= size:
+                break
+        output_path = write_optional_jsonl(str(payload.get("output_path") or ""), cases)
+        emit_success(
+            {
+                "source": source,
+                "date": date or "past-48h",
+                "scanned": len(docs),
+                "matched": len(cases),
+                "output_path": output_path,
+                "cases": [
+                    {
+                        key: compact_text(value, 1200)
+                        for key, value in case.items()
+                        if key != "raw" and value not in (None, "", [], {})
+                    }
+                    for case in cases
+                ],
+            }
+        )
+        return 0
+    except Exception as exc:  # noqa: BLE001
+        emit_error(exc)
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
+
diff --git a/skills/online-mining-v2/scripts/online_mining_common.py b/skills/online-mining-v2/scripts/online_mining_common.py
new file mode 100644
index 0000000..d4496ba
--- /dev/null
+++ b/skills/online-mining-v2/scripts/online_mining_common.py
@@ -0,0 +1,318 @@
+from __future__ import annotations
+
+"""online-mining-v2 可迁移脚本的共享逻辑。
+
+脚本只做只读 ELK 查询和轻量格式转换，输出统一 JSON，方便被不同 Agent 调用。
+"""
+
+import argparse
+import importlib.util
+import json
+import re
+import sys
+from copy import deepcopy
+from datetime import datetime, timedelta
+from pathlib import Path
+from typing import Any
+
+
+class OnlineMiningError(ValueError):
+    """线上挖掘参数或数据错误。"""
+
+
+SCRIPT_DIR = Path(__file__).resolve().parent
+SKILL_DIR = SCRIPT_DIR.parent
+REPO_ROOT = SKILL_DIR.parent.parent
+ELK_FETCH_DIR = SKILL_DIR.parent / "elk-fetch"
+
+
+SOURCE_CONFIGS: dict[str, dict[str, Any]] = {
+    "main": {
+        "profile": "default",
+        "index": "arch-flat-nlp-log-f-*",
+        "id_field": "request_id",
+        "time_field": "timestamp",
+    },
+    "pre_processing": {
+        "profile": "default",
+        "index": "pre-processing-info*",
+        "id_field": "requestId",
+        "time_field": "timestamp",
+    },
+}
+
+
+def load_json_payload(argv: list[str] | None = None) -> dict[str, Any]:
+    parser = argparse.ArgumentParser(add_help=True)
+    parser.add_argument("--input", "-i", help="JSON 参数文件；不传则从 stdin 读取")
+    args = parser.parse_args(argv)
+    text = Path(args.input).expanduser().read_text(encoding="utf-8") if args.input else sys.stdin.read()
+    if not text.strip():
+        raise OnlineMiningError("input JSON is required")
+    payload = json.loads(text)
+    if not isinstance(payload, dict):
+        raise OnlineMiningError("input JSON must be an object")
+    return payload
+
+
+def emit_success(payload: dict[str, Any]) -> None:
+    print(json.dumps({"ok": True, **payload}, ensure_ascii=False, separators=(",", ":")))
+
+
+def emit_error(exc: BaseException) -> None:
+    print(json.dumps({"ok": False, "error": str(exc)}, ensure_ascii=False, separators=(",", ":")))
+
+
+def load_elk_query_module():
+    path = ELK_FETCH_DIR / "elk_query.py"
+    if not path.exists():
+        raise OnlineMiningError(f"elk-fetch script not found: {path}")
+    spec = importlib.util.spec_from_file_location("elk_query", path)
+    if spec is None or spec.loader is None:
+        raise OnlineMiningError(f"cannot load elk_query.py from {path}")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def build_client(source: str):
+    elk = load_elk_query_module()
+    config = source_config(source)
+    profiles = elk.load_profiles()
+    return elk.build_client(profiles, str(config["profile"]))
+
+
+def source_config(source: str) -> dict[str, Any]:
+    if source not in SOURCE_CONFIGS:
+        raise OnlineMiningError(f"unknown source: {source}")
+    return SOURCE_CONFIGS[source]
+
+
+def date_range_ms(date_str: str | None) -> tuple[int, int]:
+    if date_str:
+        date_obj = datetime.strptime(str(date_str), "%Y%m%d")
+        start_ms = int(date_obj.timestamp() * 1000)
+        end_ms = int((date_obj + timedelta(days=1)).timestamp() * 1000) - 1
+    else:
+        end = datetime.now()
+        start_ms = int((end - timedelta(hours=48)).timestamp() * 1000)
+        end_ms = int(end.timestamp() * 1000)
+    return start_ms, end_ms
+
+
+def deep_parse(value: Any) -> Any:
+    if isinstance(value, str):
+        stripped = value.strip()
+        if (stripped.startswith("{") and stripped.endswith("}")) or (
+            stripped.startswith("[") and stripped.endswith("]")
+        ):
+            try:
+                return deep_parse(json.loads(stripped))
+            except Exception:
+                return value
+        return value
+    if isinstance(value, dict):
+        return {key: deep_parse(item) for key, item in value.items()}
+    if isinstance(value, list):
+        return [deep_parse(item) for item in value]
+    return value
+
+
+def get_path(obj: Any, dotted: str, default: Any = None) -> Any:
+    cur = obj
+    for part in dotted.split("."):
+        if not part:
+            continue
+        if isinstance(cur, dict):
+            cur = cur.get(part)
+        elif isinstance(cur, list):
+            try:
+                cur = cur[int(part)]
+            except (ValueError, IndexError):
+                return default
+        else:
+            return default
+    return default if cur is None else cur
+
+
+def compact_text(value: Any, limit: int = 500) -> str:
+    if value is None:
+        return ""
+    text = json.dumps(value, ensure_ascii=False) if isinstance(value, (dict, list)) else str(value)
+    text = re.sub(r"\s+", " ", text).strip()
+    return text if len(text) <= limit else text[: limit - 3] + "..."
+
+
+def as_list(value: Any) -> list[Any]:
+    if value is None:
+        return []
+    if isinstance(value, list):
+        return value
+    return [value]
+
+
+def string_values(value: Any) -> list[str]:
+    return [str(item) for item in as_list(value) if str(item).strip()]
+
+
+def match_text_filters(text: str, contains: Any = None, regex: str | None = None) -> bool:
+    for item in string_values(contains):
+        if item not in text:
+            return False
+    if regex and not re.search(regex, text):
+        return False
+    return True
+
+
+def unique_strings(items: list[Any]) -> list[str]:
+    seen: set[str] = set()
+    result: list[str] = []
+    for item in items:
+        text = str(item).strip()
+        if not text or text in seen:
+            continue
+        seen.add(text)
+        result.append(text)
+    return result
+
+
+def extract_case(source: str, doc: dict[str, Any]) -> dict[str, Any]:
+    parsed = deep_parse(doc)
+    if source == "main":
+        return extract_main_case(parsed)
+    if source == "pre_processing":
+        return extract_pre_processing_case(parsed)
+    raise OnlineMiningError(f"unknown source: {source}")
+
+
+def extract_main_case(doc: dict[str, Any]) -> dict[str, Any]:
+    device = doc.get("device")
+    return {
+        "request_id": doc.get("request_id"),
+        "timestamp": doc.get("timestamp"),
+        "query": doc.get("query"),
+        "session_id": doc.get("session_id"),
+        "device_id": doc.get("device_id"),
+        "device": device,
+        "domain": doc.get("domain"),
+        "func": doc.get("func") or doc.get("func_name"),
+        "text": doc.get("to_speak") or doc.get("text") or doc.get("display_text"),
+        "intention": doc.get("intention"),
+        "llm_agent_info": get_path(doc, "intention.intent_arbitrator_info.llm_agent_info"),
+        "score_domains": get_path(doc, "intention.intent_arbitrator_info.score_domains"),
+        "hit_rules": get_path(doc, "intention.intent_arbitrator_info.hit_rules"),
+        "raw": doc,
+    }
+
+
+def extract_pre_processing_case(doc: dict[str, Any]) -> dict[str, Any]:
+    body = doc.get("responseBoby") or {}
+    core = get_path(body, "nodes.0.core", {})
+    dispatch = get_path(core, "dispatch_large_model_result", {}) or {}
+    dispatch_input = get_path(dispatch, "dispatchLargeModelInput", {}) or {}
+    planning_debug = get_path(dispatch, "planningDebugInfo", {}) or {}
+    properties0 = get_path(body, "nodes.0.properties.0", {}) or {}
+    debug = properties0.get("debug") if isinstance(properties0, dict) else {}
+    excellent_domains = get_path(core, "excellent_domains_result", []) or []
+    domain_infos = get_path(debug, "domain_infos", []) if isinstance(debug, dict) else []
+    return {
+        "request_id": doc.get("requestId"),
+        "timestamp": doc.get("timestamp"),
+        "query": get_path(core, "query.query"),
+        "planning_result": planning_debug.get("planningOriginalResult") if isinstance(planning_debug, dict) else None,
+        "code": dispatch.get("code") if isinstance(dispatch, dict) else None,
+        "prompt_model": dispatch_input.get("promptModel") if isinstance(dispatch_input, dict) else None,
+        "hit_rules": dispatch_input.get("hitRuleList") if isinstance(dispatch_input, dict) else None,
+        "excellent_domains": excellent_domains,
+        "candidate_domains": unique_strings(
+            [get_path(item, "domain") for item in as_list(excellent_domains)]
+            + [get_path(item, "domain") for item in as_list(domain_infos)]
+        ),
+        "domain_infos": domain_infos,
+        "agent_llm_intent": properties0.get("agentLLMIntent") if isinstance(properties0, dict) else None,
+        "prompt_string": get_path(debug, "promptString") if isinstance(debug, dict) else None,
+        "planning_prompt": get_path(debug, "planningPrompt") if isinstance(debug, dict) else None,
+        "raw": doc,
+    }
+
+
+def case_matches(case: dict[str, Any], filters: dict[str, Any]) -> bool:
+    query = str(case.get("query") or "")
+    if not match_text_filters(query, filters.get("query_contains"), filters.get("query_regex")):
+        return False
+    for key in ("domain", "func", "planning_result", "code"):
+        expected = string_values(filters.get(key))
+        if expected and str(case.get(key) or "") not in expected:
+            return False
+        contains = filters.get(f"{key}_contains")
+        if contains and not match_text_filters(str(case.get(key) or ""), contains):
+            return False
+    if filters.get("prompt_contains") and not match_text_filters(
+        str(case.get("prompt_model") or case.get("prompt_string") or case.get("planning_prompt") or ""),
+        filters.get("prompt_contains"),
+    ):
+        return False
+    candidate_domains = set(string_values(case.get("candidate_domains")))
+    expected_candidates = set(string_values(filters.get("candidate_domain")))
+    if expected_candidates and not expected_candidates.intersection(candidate_domains):
+        return False
+    return True
+
+
+def search_docs(
+    *,
+    source: str,
+    date: str | None,
+    size: int,
+    query_filters: list[dict[str, Any]] | None = None,
+) -> list[dict[str, Any]]:
+    config = source_config(source)
+    client = build_client(source)
+    start_ms, end_ms = date_range_ms(date)
+    filters = [{"range": {str(config["time_field"]): {"gte": start_ms, "lte": end_ms}}}]
+    filters.extend(deepcopy(query_filters or []))
+    body = {
+        "query": {"bool": {"filter": filters}},
+        "size": size,
+        "sort": [{str(config["time_field"]): {"order": "desc"}}],
+    }
+    response = client.search(index=str(config["index"]), body=body)
+    return [deep_parse(hit.get("_source", {})) for hit in response.get("hits", {}).get("hits", [])]
+
+
+def fetch_one_by_request_id(source: str, request_id: str, date: str | None = None) -> dict[str, Any] | None:
+    config = source_config(source)
+    docs = search_docs(
+        source=source,
+        date=date,
+        size=1,
+        query_filters=[{"term": {str(config["id_field"]): request_id}}],
+    )
+    if not docs:
+        # 部分 request id 会带后缀，兜底用 wildcard。
+        docs = search_docs(
+            source=source,
+            date=date,
+            size=1,
+            query_filters=[{"wildcard": {str(config["id_field"]): {"value": f"{request_id}*"}}}],
+        )
+    return docs[0] if docs else None
+
+
+def write_optional_jsonl(path: str | None, rows: list[dict[str, Any]]) -> str | None:
+    if not path:
+        return None
+    output = resolve_portable_path(path)
+    output.parent.mkdir(parents=True, exist_ok=True)
+    with output.open("w", encoding="utf-8") as fh:
+        for row in rows:
+            fh.write(json.dumps(row, ensure_ascii=False, separators=(",", ":")) + "\n")
+    return str(output)
+
+
+def resolve_portable_path(path: str) -> Path:
+    raw = Path(path).expanduser()
+    if raw.is_absolute():
+        return raw
+    return (REPO_ROOT / raw).resolve()
+