Add router session mining tools

src/agent_tools.py

@@ -35,6 +35,13 @@ from .data_agent_inputs import (
     load_input_sources,
     render_source_context,
 )
+from .data_agent_router_sessions import (
+    DataAgentRouterSessionError,
+    convert_router_candidates_to_records,
+    profile_router_sessions,
+    sample_router_candidates,
+    search_router_sessions,
+)
 from .session_env_vars import get_session_env_vars
 
 if TYPE_CHECKING:
@@ -1466,6 +1473,137 @@ def default_tool_registry() -> dict[str, AgentTool]:
             },
             handler=_data_agent_render_source_context_tool,
         ),
+        AgentTool(
+            name='data_agent_profile_router_sessions',
+            description=(
+                'Profile local router_session_parquet date partitions before mining online sessions. '
+                'Returns schema, sampled row count, distributions, and example sessions.'
+            ),
+            parameters={
+                'type': 'object',
+                'properties': {
+                    'dates': {
+                        'type': 'array',
+                        'items': {'type': 'string'},
+                        'description': 'Date partitions such as ["20260428"]. Resolved under router_session_parquet/date=YYYYMMDD.',
+                    },
+                    'paths': {
+                        'type': 'array',
+                        'items': {'type': 'string'},
+                        'description': 'Workspace-relative parquet files or date directories. Use when dates is not enough.',
+                    },
+                    'max_files': {'type': 'integer', 'minimum': 1, 'maximum': 200},
+                    'max_rows_per_file': {'type': 'integer', 'minimum': 1, 'maximum': 100000},
+                    'examples_limit': {'type': 'integer', 'minimum': 0, 'maximum': 20},
+                },
+            },
+            handler=_data_agent_profile_router_sessions_tool,
+        ),
+        AgentTool(
+            name='data_agent_search_router_sessions',
+            description=(
+                'Search local router_session_parquet data with reviewable filters such as device, domain, intent, func, '
+                'query keywords, regex, turn count, and match scope. Returns turn-level candidates with prev turns.'
+            ),
+            parameters={
+                'type': 'object',
+                'properties': {
+                    'dates': {
+                        'type': 'array',
+                        'items': {'type': 'string'},
+                        'description': 'Date partitions such as ["20260428"]. Resolved under router_session_parquet/date=YYYYMMDD.',
+                    },
+                    'paths': {
+                        'type': 'array',
+                        'items': {'type': 'string'},
+                        'description': 'Workspace-relative parquet files or date directories. Use when dates is not enough.',
+                    },
+                    'devices': {'type': 'array', 'items': {'type': 'string'}},
+                    'domains': {'type': 'array', 'items': {'type': 'string'}},
+                    'intents': {'type': 'array', 'items': {'type': 'string'}},
+                    'funcs': {'type': 'array', 'items': {'type': 'string'}},
+                    'query_keywords': {'type': 'array', 'items': {'type': 'string'}},
+                    'keyword_match_mode': {'type': 'string', 'enum': ['any', 'all']},
+                    'query_regex': {'type': 'string'},
+                    'match_scope': {'type': 'string', 'enum': ['any_turn', 'last_turn']},
+                    'turn_count_min': {'type': 'integer', 'minimum': 1},
+                    'turn_count_max': {'type': 'integer', 'minimum': 1},
+                    'max_files': {'type': 'integer', 'minimum': 1, 'maximum': 200},
+                    'max_rows_per_file': {'type': 'integer', 'minimum': 1, 'maximum': 100000},
+                    'max_candidates': {'type': 'integer', 'minimum': 1, 'maximum': 5000},
+                },
+            },
+            handler=_data_agent_search_router_sessions_tool,
+        ),
+        AgentTool(
+            name='data_agent_sample_router_candidates',
+            description=(
+                'Sample router session candidates for human review and return candidate-level summary statistics. '
+                'Use after data_agent_search_router_sessions.'
+            ),
+            parameters={
+                'type': 'object',
+                'properties': {
+                    'candidates': {
+                        'oneOf': [
+                            {'type': 'array'},
+                            {'type': 'object'},
+                            {'type': 'string'},
+                        ],
+                        'description': 'Candidates array, search result object, or JSON string containing candidates.',
+                    },
+                    'sample_size': {'type': 'integer', 'minimum': 1, 'maximum': 1000},
+                    'strategy': {'type': 'string', 'enum': ['random', 'first', 'stride']},
+                    'seed': {'type': 'integer'},
+                },
+                'required': ['candidates'],
+            },
+            handler=_data_agent_sample_router_candidates_tool,
+        ),
+        AgentTool(
+            name='data_agent_convert_router_candidates_to_records',
+            description=(
+                'Convert reviewed online router session candidates directly into canonical data-agent records. '
+                'Use this when the user wants to keep mined online data as samples; do not use generation-plan tools for this path.'
+            ),
+            parameters={
+                'type': 'object',
+                'properties': {
+                    'candidates': {
+                        'oneOf': [
+                            {'type': 'array'},
+                            {'type': 'object'},
+                            {'type': 'string'},
+                        ],
+                        'description': 'Candidates array, search/sample result object, or JSON string containing candidates.',
+                    },
+                    'dataset_label': {'type': 'string'},
+                    'default_target': {
+                        'type': 'string',
+                        'description': 'Target label used for included candidates unless review_decisions provides a target.',
+                    },
+                    'review_decisions': {
+                        'type': 'array',
+                        'description': 'Optional include/exclude/uncertain decisions keyed by semantic_session_id or req_id.',
+                        'items': {
+                            'type': 'object',
+                            'properties': {
+                                'semantic_session_id': {'type': 'string'},
+                                'req_id': {'type': 'string'},
+                                'matched_turn_index': {'type': 'integer'},
+                                'decision': {'type': 'string', 'enum': ['include', 'exclude', 'uncertain']},
+                                'target': {'type': 'string'},
+                                'notes': {'type': 'string'},
+                            },
+                        },
+                    },
+                    'batch_id': {'type': 'string'},
+                    'include_uncertain': {'type': 'boolean'},
+                },
+                'required': ['candidates', 'dataset_label'],
+            },
+            handler=_data_agent_convert_router_candidates_to_records_tool,
+        ),
         AgentTool(
             name='data_agent_show_generation_plan',
             description='Show a data-agent generation plan for human review.',
@@ -1689,6 +1827,15 @@ def _coerce_int(arguments: dict[str, Any], key: str, default: int) -> int:
     return value
 
 
+def _optional_int(arguments: dict[str, Any], key: str) -> int | None:
+    value = arguments.get(key)
+    if value is None:
+        return None
+    if isinstance(value, bool) or not isinstance(value, int):
+        raise ToolExecutionError(f'{key} must be an integer')
+    return value
+
+
 def _coerce_float(arguments: dict[str, Any], key: str, default: float) -> float:
     value = arguments.get(key, default)
     if isinstance(value, bool) or not isinstance(value, (int, float)):
@@ -1855,6 +2002,88 @@ def _data_agent_render_source_context_tool(arguments: dict[str, Any], context: T
     return data_agent_dumps_payload(payload)
 
 
+def _data_agent_profile_router_sessions_tool(arguments: dict[str, Any], context: ToolExecutionContext) -> str:
+    try:
+        payload = profile_router_sessions(
+            context.root,
+            dates=_optional_string_list(arguments, 'dates'),
+            paths=_optional_string_list(arguments, 'paths'),
+            max_files=_coerce_int(arguments, 'max_files', 8),
+            max_rows_per_file=_coerce_int(arguments, 'max_rows_per_file', 2000),
+            examples_limit=_coerce_int(arguments, 'examples_limit', 3),
+        )
+    except DataAgentRouterSessionError as exc:
+        raise ToolExecutionError(str(exc)) from exc
+    return data_agent_dumps_payload(payload)
+
+
+def _data_agent_search_router_sessions_tool(arguments: dict[str, Any], context: ToolExecutionContext) -> str:
+    try:
+        payload = search_router_sessions(
+            context.root,
+            dates=_optional_string_list(arguments, 'dates'),
+            paths=_optional_string_list(arguments, 'paths'),
+            devices=_optional_string_list(arguments, 'devices'),
+            domains=_optional_string_list(arguments, 'domains'),
+            intents=_optional_string_list(arguments, 'intents'),
+            funcs=_optional_string_list(arguments, 'funcs'),
+            query_keywords=_optional_string_list(arguments, 'query_keywords'),
+            keyword_match_mode=_optional_string(arguments, 'keyword_match_mode') or 'any',
+            query_regex=_optional_string(arguments, 'query_regex'),
+            match_scope=_optional_string(arguments, 'match_scope') or 'any_turn',
+            turn_count_min=_optional_int(arguments, 'turn_count_min'),
+            turn_count_max=_optional_int(arguments, 'turn_count_max'),
+            max_files=_coerce_int(arguments, 'max_files', 20),
+            max_rows_per_file=_coerce_int(arguments, 'max_rows_per_file', 5000),
+            max_candidates=_coerce_int(arguments, 'max_candidates', 200),
+        )
+    except DataAgentRouterSessionError as exc:
+        raise ToolExecutionError(str(exc)) from exc
+    return data_agent_dumps_payload(payload)
+
+
+def _data_agent_sample_router_candidates_tool(arguments: dict[str, Any], _context: ToolExecutionContext) -> str:
+    if 'candidates' not in arguments:
+        raise ToolExecutionError('candidates is required')
+    try:
+        payload = sample_router_candidates(
+            arguments['candidates'],
+            sample_size=_coerce_int(arguments, 'sample_size', 100),
+            strategy=_optional_string(arguments, 'strategy') or 'random',
+            seed=_coerce_int(arguments, 'seed', 20260430),
+        )
+    except DataAgentRouterSessionError as exc:
+        raise ToolExecutionError(str(exc)) from exc
+    return data_agent_dumps_payload(payload)
+
+
+def _data_agent_convert_router_candidates_to_records_tool(
+    arguments: dict[str, Any],
+    _context: ToolExecutionContext,
+) -> str:
+    if 'candidates' not in arguments:
+        raise ToolExecutionError('candidates is required')
+    review_decisions = arguments.get('review_decisions')
+    if review_decisions is not None and not isinstance(review_decisions, list):
+        raise ToolExecutionError('review_decisions must be an array')
+    include_uncertain = arguments.get('include_uncertain', False)
+    if not isinstance(include_uncertain, bool):
+        raise ToolExecutionError('include_uncertain must be a boolean')
+    try:
+        payload = convert_router_candidates_to_records(
+            arguments['candidates'],
+            dataset_label=_require_string(arguments, 'dataset_label'),
+            default_target=_optional_string(arguments, 'default_target'),
+            review_decisions=review_decisions,
+            batch_id=_optional_string(arguments, 'batch_id') or 'router',
+            include_uncertain=include_uncertain,
+        )
+        payload['validation'] = validate_dataset_records(payload['records'])
+    except (DataAgentRouterSessionError, DataRecordError) as exc:
+        raise ToolExecutionError(str(exc)) from exc
+    return data_agent_dumps_payload(payload)
+
+
 def _data_agent_show_generation_plan_tool(arguments: dict[str, Any], context: ToolExecutionContext) -> str:
     try:
         payload = get_generation_plan(

src/data_agent_records.py

@@ -402,8 +402,8 @@ def validate_dataset_records(records: list[dict[str, Any]]) -> dict[str, Any]:
                     continue
                 if not _non_empty_str(item.get('query')):
                     errors.append(_issue(f'{item_prefix}.query', 'query is required'))
-                if not _non_empty_str(item.get('tts')):
+                if 'tts' not in item or not isinstance(item.get('tts'), str):
-                    errors.append(_issue(f'{item_prefix}.tts', 'tts is required'))
+                    errors.append(_issue(f'{item_prefix}.tts', 'tts must be a string'))
                 timestamp = item.get('timestamp')
                 if not _int_like(timestamp):
                     errors.append(_issue(f'{item_prefix}.timestamp', 'timestamp must be an integer'))

src/data_agent_router_sessions.py

@@ -0,0 +1,739 @@
+from __future__ import annotations
+
+import json
+import random
+import re
+from collections import Counter
+from pathlib import Path
+from typing import Any, Iterable
+
+
+class DataAgentRouterSessionError(ValueError):
+    """路由 session parquet 工具的输入或运行错误。"""
+
+
+DEFAULT_ROUTER_SESSION_ROOT = 'router_session_parquet'
+
+
+def profile_router_sessions(
+    root: str | Path,
+    *,
+    dates: list[str] | None = None,
+    paths: list[str] | None = None,
+    max_files: int = 8,
+    max_rows_per_file: int = 2000,
+    examples_limit: int = 3,
+) -> dict[str, Any]:
+    """读取路由 session parquet 的小样本概貌，辅助后续制定挖掘策略。"""
+
+    files = _resolve_parquet_files(root, dates=dates, paths=paths, max_files=max_files)
+    rows_seen = 0
+    missing_req_id = 0
+    devices: Counter[str] = Counter()
+    turn_counts: Counter[str] = Counter()
+    domains: Counter[str] = Counter()
+    intents: Counter[str] = Counter()
+    funcs: Counter[str] = Counter()
+    action_keys: Counter[str] = Counter()
+    examples: list[dict[str, Any]] = []
+    schema = ''
+
+    for parquet_file in files:
+        parquet = _import_pyarrow_parquet()
+        if not schema:
+            schema = str(parquet.ParquetFile(parquet_file).schema)
+        for row in _iter_parquet_rows(parquet_file, max_rows_per_file=max_rows_per_file):
+            rows_seen += 1
+            session = _normalize_session(row, parquet_file)
+            if not session.get('req_id'):
+                missing_req_id += 1
+            devices[_counter_key(session.get('device'))] += 1
+            turn_counts[str(session.get('turn_count', 0))] += 1
+            for turn in session.get('turns', []):
+                action = turn.get('action') or {}
+                for key in action:
+                    action_keys[key] += 1
+                domains[_counter_key(action.get('domain'))] += 1
+                intents[_counter_key(action.get('intent'))] += 1
+                if action.get('func') is not None:
+                    funcs[_counter_key(action.get('func'))] += 1
+            if len(examples) < examples_limit:
+                examples.append(_session_example(session))
+
+    return {
+        'ok': True,
+        'action': 'data_agent_profile_router_sessions',
+        'source': {
+            'dates': dates or [],
+            'paths': [str(item) for item in files],
+            'file_count': len(files),
+            'max_rows_per_file': max_rows_per_file,
+        },
+        'schema': schema,
+        'sampled_row_count': rows_seen,
+        'missing_req_id_count': missing_req_id,
+        'distributions': {
+            'device': _top_counts(devices),
+            'turn_count': _top_counts(turn_counts),
+            'domain': _top_counts(domains),
+            'intent': _top_counts(intents),
+            'func': _top_counts(funcs),
+            'action_json_keys': _top_counts(action_keys),
+        },
+        'examples': examples,
+    }
+
+
+def search_router_sessions(
+    root: str | Path,
+    *,
+    dates: list[str] | None = None,
+    paths: list[str] | None = None,
+    devices: list[str] | None = None,
+    domains: list[str] | None = None,
+    intents: list[str] | None = None,
+    funcs: list[str] | None = None,
+    query_keywords: list[str] | None = None,
+    keyword_match_mode: str = 'any',
+    query_regex: str = '',
+    match_scope: str = 'any_turn',
+    turn_count_min: int | None = None,
+    turn_count_max: int | None = None,
+    max_files: int = 20,
+    max_rows_per_file: int = 5000,
+    max_candidates: int = 200,
+) -> dict[str, Any]:
+    """按可 review 的策略召回路由 session 候选，输出 turn 级候选。"""
+
+    if match_scope not in {'any_turn', 'last_turn'}:
+        raise DataAgentRouterSessionError('match_scope must be any_turn or last_turn')
+    if keyword_match_mode not in {'any', 'all'}:
+        raise DataAgentRouterSessionError('keyword_match_mode must be any or all')
+    regex = _compile_regex(query_regex)
+    files = _resolve_parquet_files(root, dates=dates, paths=paths, max_files=max_files)
+    candidate_rows: list[dict[str, Any]] = []
+    scanned_rows = 0
+    scanned_turns = 0
+
+    filters = {
+        'devices': devices or [],
+        'domains': domains or [],
+        'intents': intents or [],
+        'funcs': funcs or [],
+        'query_keywords': query_keywords or [],
+        'keyword_match_mode': keyword_match_mode,
+        'query_regex': query_regex,
+        'match_scope': match_scope,
+        'turn_count_min': turn_count_min,
+        'turn_count_max': turn_count_max,
+    }
+
+    for parquet_file in files:
+        for row in _iter_parquet_rows(parquet_file, max_rows_per_file=max_rows_per_file):
+            scanned_rows += 1
+            session = _normalize_session(row, parquet_file)
+            if not _session_matches(session, devices, turn_count_min, turn_count_max):
+                continue
+            turn_indexes = _candidate_turn_indexes(session.get('turns', []), match_scope)
+            for turn_index in turn_indexes:
+                scanned_turns += 1
+                turn = session['turns'][turn_index]
+                if not _turn_matches(
+                    turn,
+                    domains=domains,
+                    intents=intents,
+                    funcs=funcs,
+                    query_keywords=query_keywords,
+                    keyword_match_mode=keyword_match_mode,
+                    regex=regex,
+                ):
+                    continue
+                candidate_rows.append(_candidate_from_session(session, turn_index))
+                if len(candidate_rows) >= max_candidates:
+                    return _search_payload(files, filters, scanned_rows, scanned_turns, candidate_rows, truncated=True)
+
+    return _search_payload(files, filters, scanned_rows, scanned_turns, candidate_rows, truncated=False)
+
+
+def sample_router_candidates(
+    candidates: list[dict[str, Any]] | dict[str, Any] | str,
+    *,
+    sample_size: int = 100,
+    strategy: str = 'random',
+    seed: int = 20260430,
+) -> dict[str, Any]:
+    """从候选集中做稳定抽样，并给出用于人工 review 的基础统计。"""
+
+    candidate_rows = _extract_candidates(candidates)
+    if sample_size < 1:
+        raise DataAgentRouterSessionError('sample_size must be greater than 0')
+    if strategy not in {'random', 'first', 'stride'}:
+        raise DataAgentRouterSessionError('strategy must be random, first, or stride')
+
+    if strategy == 'first':
+        sampled = candidate_rows[:sample_size]
+    elif strategy == 'stride':
+        sampled = _stride_sample(candidate_rows, sample_size)
+    else:
+        sampled = list(candidate_rows)
+        random.Random(seed).shuffle(sampled)
+        sampled = sampled[:sample_size]
+
+    return {
+        'ok': True,
+        'action': 'data_agent_sample_router_candidates',
+        'total_candidate_count': len(candidate_rows),
+        'sampled_count': len(sampled),
+        'strategy': strategy,
+        'seed': seed if strategy == 'random' else None,
+        'summary': _candidate_summary(candidate_rows),
+        'sample_summary': _candidate_summary(sampled),
+        'candidates': sampled,
+    }
+
+
+def convert_router_candidates_to_records(
+    candidates: list[dict[str, Any]] | dict[str, Any] | str,
+    *,
+    dataset_label: str,
+    default_target: str = '',
+    review_decisions: list[dict[str, Any]] | None = None,
+    batch_id: str = 'router',
+    include_uncertain: bool = False,
+) -> dict[str, Any]:
+    """把 review 后的线上候选直接转换为 canonical records，不经过数据生成流程。"""
+
+    if not dataset_label.strip():
+        raise DataAgentRouterSessionError('dataset_label is required')
+    if not batch_id.strip():
+        raise DataAgentRouterSessionError('batch_id is required')
+
+    candidate_rows = _extract_candidates(candidates)
+    decisions = _normalize_review_decisions(review_decisions or [])
+    records: list[dict[str, Any]] = []
+    skipped: list[dict[str, Any]] = []
+
+    for index, candidate in enumerate(candidate_rows, 1):
+        decision = _decision_for_candidate(decisions, candidate)
+        status = str(decision.get('decision') or 'include').strip().lower()
+        if status in {'exclude', 'drop', 'reject'}:
+            skipped.append(_skip_item(candidate, status, decision.get('notes')))
+            continue
+        if status in {'uncertain', 'unknown', '不确定'} and not include_uncertain:
+            skipped.append(_skip_item(candidate, status, decision.get('notes')))
+            continue
+
+        target = str(decision.get('target') or default_target).strip()
+        if not target:
+            raise DataAgentRouterSessionError(
+                f'target is required for candidate {candidate.get("semantic_session_id")}#{candidate.get("matched_turn_index")}'
+            )
+        records.append(
+            _router_candidate_to_record(
+                candidate,
+                dataset_label=str(decision.get('dataset_label') or dataset_label).strip(),
+                target=target,
+                notes=str(decision.get('notes') or '').strip(),
+                batch_id=batch_id,
+                index=len(records) + 1,
+            )
+        )
+
+    return {
+        'ok': True,
+        'action': 'data_agent_convert_router_candidates_to_records',
+        'input_candidate_count': len(candidate_rows),
+        'record_count': len(records),
+        'skipped_count': len(skipped),
+        'skipped': skipped,
+        'records': records,
+    }
+
+
+def _resolve_parquet_files(
+    root: str | Path,
+    *,
+    dates: list[str] | None,
+    paths: list[str] | None,
+    max_files: int,
+) -> list[Path]:
+    if max_files < 1:
+        raise DataAgentRouterSessionError('max_files must be greater than 0')
+    root_path = Path(root).resolve()
+    requested: list[Path] = []
+    if dates:
+        for date in dates:
+            if not re.fullmatch(r'\d{8}', date):
+                raise DataAgentRouterSessionError(f'invalid date: {date}')
+            requested.append(root_path / DEFAULT_ROUTER_SESSION_ROOT / f'date={date}')
+    if paths:
+        for raw_path in paths:
+            if not isinstance(raw_path, str) or not raw_path.strip():
+                raise DataAgentRouterSessionError('paths items must be non-empty strings')
+            expanded = Path(raw_path).expanduser()
+            requested.append(expanded if expanded.is_absolute() else root_path / expanded)
+    if not requested:
+        raise DataAgentRouterSessionError('dates or paths is required')
+
+    files: list[Path] = []
+    for path in requested:
+        resolved = path.resolve()
+        try:
+            resolved.relative_to(root_path)
+        except ValueError as exc:
+            raise DataAgentRouterSessionError(f'path escapes workspace root: {path}') from exc
+        if not resolved.exists():
+            raise DataAgentRouterSessionError(f'path not found: {path}')
+        if resolved.is_dir():
+            files.extend(sorted(item for item in resolved.glob('*.parquet') if item.is_file()))
+        elif resolved.suffix == '.parquet':
+            files.append(resolved)
+        else:
+            raise DataAgentRouterSessionError(f'path is not a parquet file or directory: {path}')
+
+    deduped = []
+    seen: set[Path] = set()
+    for file_path in files:
+        if file_path in seen:
+            continue
+        seen.add(file_path)
+        deduped.append(file_path)
+    if not deduped:
+        raise DataAgentRouterSessionError('no parquet files found')
+    return deduped[:max_files]
+
+
+def _iter_parquet_rows(path: Path, *, max_rows_per_file: int) -> Iterable[dict[str, Any]]:
+    if max_rows_per_file < 1:
+        raise DataAgentRouterSessionError('max_rows_per_file must be greater than 0')
+    parquet = _import_pyarrow_parquet()
+    parquet_file = parquet.ParquetFile(path)
+    emitted = 0
+    batch_size = min(1024, max_rows_per_file)
+    for batch in parquet_file.iter_batches(batch_size=batch_size):
+        for row in batch.to_pylist():
+            yield row
+            emitted += 1
+            if emitted >= max_rows_per_file:
+                return
+
+
+def _import_pyarrow_parquet() -> Any:
+    try:
+        import pyarrow.parquet as parquet
+    except ImportError as exc:
+        raise DataAgentRouterSessionError('读取 router session parquet 需要安装 pyarrow') from exc
+    return parquet
+
+
+def _normalize_session(row: dict[str, Any], source_path: Path) -> dict[str, Any]:
+    turns = row.get('turns_array') or []
+    normalized_turns = []
+    for turn in turns:
+        if not isinstance(turn, dict):
+            continue
+        action, parse_error = _parse_action(turn.get('action_json'))
+        normalized_turns.append(
+            {
+                'timestamp': turn.get('timestamp'),
+                'query': str(turn.get('query') or ''),
+                'tts': str(turn.get('tts') or ''),
+                'action': action,
+                'action_parse_error': parse_error,
+            }
+        )
+    return {
+        'semantic_session_id': row.get('semantic_session_id'),
+        'device': row.get('device'),
+        'req_id': row.get('req_id'),
+        'turn_count': int(row.get('turn_count') or len(normalized_turns)),
+        'turns': normalized_turns,
+        'source_path': str(source_path),
+        'date': _date_from_path(source_path),
+    }
+
+
+def _parse_action(raw: Any) -> tuple[dict[str, Any], str | None]:
+    if not raw:
+        return {}, None
+    if isinstance(raw, dict):
+        return raw, None
+    if not isinstance(raw, str):
+        return {}, f'unsupported action_json type: {type(raw).__name__}'
+    try:
+        payload = json.loads(raw)
+    except json.JSONDecodeError as exc:
+        return {}, str(exc)
+    if not isinstance(payload, dict):
+        return {}, 'action_json is not an object'
+    return payload, None
+
+
+def _date_from_path(path: Path) -> str:
+    for part in path.parts:
+        if part.startswith('date='):
+            return part.split('=', 1)[1]
+    return ''
+
+
+def _session_matches(
+    session: dict[str, Any],
+    devices: list[str] | None,
+    turn_count_min: int | None,
+    turn_count_max: int | None,
+) -> bool:
+    if devices and session.get('device') not in set(devices):
+        return False
+    turn_count = int(session.get('turn_count') or 0)
+    if turn_count_min is not None and turn_count < turn_count_min:
+        return False
+    if turn_count_max is not None and turn_count > turn_count_max:
+        return False
+    return True
+
+
+def _turn_matches(
+    turn: dict[str, Any],
+    *,
+    domains: list[str] | None,
+    intents: list[str] | None,
+    funcs: list[str] | None,
+    query_keywords: list[str] | None,
+    keyword_match_mode: str,
+    regex: re.Pattern[str] | None,
+) -> bool:
+    action = turn.get('action') or {}
+    if domains and action.get('domain') not in set(domains):
+        return False
+    if intents and action.get('intent') not in set(intents):
+        return False
+    if funcs and action.get('func') not in set(funcs):
+        return False
+    query = str(turn.get('query') or '')
+    keywords = [keyword for keyword in query_keywords or [] if keyword]
+    if keywords:
+        hits = [keyword in query for keyword in keywords]
+        if keyword_match_mode == 'all' and not all(hits):
+            return False
+        if keyword_match_mode == 'any' and not any(hits):
+            return False
+    if regex and not regex.search(query):
+        return False
+    return True
+
+
+def _compile_regex(pattern: str) -> re.Pattern[str] | None:
+    if not pattern:
+        return None
+    try:
+        return re.compile(pattern)
+    except re.error as exc:
+        raise DataAgentRouterSessionError(f'invalid query_regex: {exc}') from exc
+
+
+def _candidate_turn_indexes(turns: list[dict[str, Any]], match_scope: str) -> list[int]:
+    if not turns:
+        return []
+    if match_scope == 'last_turn':
+        return [len(turns) - 1]
+    return list(range(len(turns)))
+
+
+def _candidate_from_session(session: dict[str, Any], turn_index: int) -> dict[str, Any]:
+    turns = session.get('turns') or []
+    turn = turns[turn_index]
+    return {
+        'semantic_session_id': session.get('semantic_session_id'),
+        'req_id': session.get('req_id'),
+        'device': session.get('device'),
+        'turn_count': session.get('turn_count'),
+        'matched_turn_index': turn_index,
+        'matched_turn': turn,
+        'prev_turns': [
+            {
+                'timestamp': prev.get('timestamp'),
+                'query': prev.get('query'),
+                'tts': prev.get('tts'),
+                'action': prev.get('action'),
+            }
+            for prev in turns[max(0, turn_index - 10) : turn_index]
+        ],
+        'source_path': session.get('source_path'),
+        'date': session.get('date'),
+    }
+
+
+def _search_payload(
+    files: list[Path],
+    filters: dict[str, Any],
+    scanned_rows: int,
+    scanned_turns: int,
+    candidates: list[dict[str, Any]],
+    *,
+    truncated: bool,
+) -> dict[str, Any]:
+    return {
+        'ok': True,
+        'action': 'data_agent_search_router_sessions',
+        'source': {
+            'paths': [str(item) for item in files],
+            'file_count': len(files),
+        },
+        'filters': filters,
+        'scanned_row_count': scanned_rows,
+        'scanned_turn_count': scanned_turns,
+        'candidate_count': len(candidates),
+        'truncated': truncated,
+        'summary': _candidate_summary(candidates),
+        'candidates': candidates,
+    }
+
+
+def _extract_candidates(candidates: list[dict[str, Any]] | dict[str, Any] | str) -> list[dict[str, Any]]:
+    payload = candidates
+    if isinstance(payload, str):
+        try:
+            payload = json.loads(payload)
+        except json.JSONDecodeError as exc:
+            raise DataAgentRouterSessionError(f'candidates is not valid JSON: {exc}') from exc
+    if isinstance(payload, dict):
+        payload = payload.get('candidates')
+    if not isinstance(payload, list):
+        raise DataAgentRouterSessionError('candidates must be an array or an object with candidates')
+    for index, item in enumerate(payload):
+        if not isinstance(item, dict):
+            raise DataAgentRouterSessionError(f'candidates[{index}] must be an object')
+    return payload
+
+
+def _normalize_review_decisions(review_decisions: list[dict[str, Any]]) -> dict[tuple[str, str], dict[str, Any]]:
+    decisions: dict[tuple[str, str], dict[str, Any]] = {}
+    if not isinstance(review_decisions, list):
+        raise DataAgentRouterSessionError('review_decisions must be an array')
+    for index, item in enumerate(review_decisions):
+        if not isinstance(item, dict):
+            raise DataAgentRouterSessionError(f'review_decisions[{index}] must be an object')
+        key = _decision_key_from_item(item)
+        if key == ('', ''):
+            raise DataAgentRouterSessionError(
+                f'review_decisions[{index}] must include semantic_session_id or req_id'
+            )
+        decisions[key] = item
+    return decisions
+
+
+def _decision_key_from_item(item: dict[str, Any]) -> tuple[str, str]:
+    session_id = str(item.get('semantic_session_id') or item.get('session_id') or item.get('req_id') or '').strip()
+    turn_index = item.get('matched_turn_index')
+    return session_id, '' if turn_index is None else str(turn_index)
+
+
+def _candidate_key(candidate: dict[str, Any]) -> tuple[str, str]:
+    session_id = str(candidate.get('semantic_session_id') or candidate.get('req_id') or '').strip()
+    turn_index = candidate.get('matched_turn_index')
+    exact_key = (session_id, '' if turn_index is None else str(turn_index))
+    if exact_key in {('', '')}:
+        return exact_key
+    return exact_key
+
+
+def _decision_for_candidate(
+    decisions: dict[tuple[str, str], dict[str, Any]],
+    candidate: dict[str, Any],
+) -> dict[str, Any]:
+    turn_index = candidate.get('matched_turn_index')
+    turn_key = '' if turn_index is None else str(turn_index)
+    keys = [
+        (str(candidate.get('semantic_session_id') or '').strip(), turn_key),
+        (str(candidate.get('semantic_session_id') or '').strip(), ''),
+        (str(candidate.get('req_id') or '').strip(), turn_key),
+        (str(candidate.get('req_id') or '').strip(), ''),
+    ]
+    for key in keys:
+        if key in decisions:
+            return decisions[key]
+    return {}
+
+
+def _router_candidate_to_record(
+    candidate: dict[str, Any],
+    *,
+    dataset_label: str,
+    target: str,
+    notes: str,
+    batch_id: str,
+    index: int,
+) -> dict[str, Any]:
+    matched_turn = candidate.get('matched_turn')
+    if not isinstance(matched_turn, dict):
+        raise DataAgentRouterSessionError(f'candidate {index} matched_turn is required')
+    query = str(matched_turn.get('query') or '').strip()
+    if not query:
+        raise DataAgentRouterSessionError(f'candidate {index} matched_turn.query is required')
+    timestamp = _int_or_zero(matched_turn.get('timestamp'))
+    if timestamp <= 0:
+        raise DataAgentRouterSessionError(f'candidate {index} matched_turn.timestamp is required')
+    request_id = str(candidate.get('req_id') or '').strip()
+    if not request_id:
+        raise DataAgentRouterSessionError(f'candidate {index} req_id is required')
+
+    action = matched_turn.get('action') if isinstance(matched_turn.get('action'), dict) else {}
+    prev_session = _prev_turns_to_prev_session(candidate.get('prev_turns'))
+    return {
+        'record_id': f'online_{_safe_token(batch_id)}_{_safe_token(request_id)}_{index:06d}',
+        'source': {
+            'type': 'online',
+            'request_id': request_id,
+            'timestamp': timestamp,
+        },
+        'turn': {
+            'query': query,
+            'timestamp': timestamp,
+        },
+        'prev_session': prev_session,
+        'context': {
+            'semantic_session_id': candidate.get('semantic_session_id'),
+            'device': candidate.get('device'),
+            'date': candidate.get('date'),
+            'turn_count': candidate.get('turn_count'),
+            'matched_turn_index': candidate.get('matched_turn_index'),
+            'source_path': candidate.get('source_path'),
+            'domain': action.get('domain'),
+            'intent': action.get('intent'),
+            'func': action.get('func'),
+            'norm_code': action.get('norm_code'),
+        },
+        'label': {
+            'dataset_label': dataset_label,
+            'target': target,
+            'target_type': _target_type(target),
+        },
+        'meta': {
+            'case_name': '',
+            'notes': notes,
+        },
+    }
+
+
+def _prev_turns_to_prev_session(value: Any) -> list[dict[str, Any]]:
+    if not isinstance(value, list):
+        return []
+    prev_session: list[dict[str, Any]] = []
+    for item in value[-10:]:
+        if not isinstance(item, dict):
+            continue
+        query = str(item.get('query') or '').strip()
+        if not query:
+            continue
+        prev_session.append(
+            {
+                'query': query,
+                'tts': str(item.get('tts') or ''),
+                'timestamp': _int_or_zero(item.get('timestamp')),
+            }
+        )
+    return prev_session
+
+
+def _skip_item(candidate: dict[str, Any], decision: str, notes: Any) -> dict[str, Any]:
+    return {
+        'semantic_session_id': candidate.get('semantic_session_id'),
+        'req_id': candidate.get('req_id'),
+        'matched_turn_index': candidate.get('matched_turn_index'),
+        'decision': decision,
+        'notes': str(notes or ''),
+    }
+
+
+def _target_type(target: str) -> str:
+    if target.startswith('Agent('):
+        return 'agent'
+    if target:
+        return 'function'
+    return 'unknown'
+
+
+def _safe_token(value: str) -> str:
+    token = re.sub(r'[^A-Za-z0-9_.-]+', '_', value.strip())
+    return token.strip('_.-') or 'item'
+
+
+def _int_or_zero(value: Any) -> int:
+    if isinstance(value, bool):
+        return 0
+    if isinstance(value, int):
+        return value
+    if isinstance(value, float) and value.is_integer():
+        return int(value)
+    if isinstance(value, str) and value.isdigit():
+        return int(value)
+    return 0
+
+
+def _candidate_summary(candidates: list[dict[str, Any]]) -> dict[str, Any]:
+    devices: Counter[str] = Counter()
+    turn_counts: Counter[str] = Counter()
+    domains: Counter[str] = Counter()
+    intents: Counter[str] = Counter()
+    funcs: Counter[str] = Counter()
+    top_queries: Counter[str] = Counter()
+    for item in candidates:
+        devices[_counter_key(item.get('device'))] += 1
+        turn_counts[str(item.get('turn_count') or 0)] += 1
+        matched_turn = item.get('matched_turn') or {}
+        if isinstance(matched_turn, dict):
+            query = str(matched_turn.get('query') or '')
+            if query:
+                top_queries[query] += 1
+            action = matched_turn.get('action') or {}
+            if isinstance(action, dict):
+                domains[_counter_key(action.get('domain'))] += 1
+                intents[_counter_key(action.get('intent'))] += 1
+                if action.get('func') is not None:
+                    funcs[_counter_key(action.get('func'))] += 1
+    return {
+        'device': _top_counts(devices),
+        'turn_count': _top_counts(turn_counts),
+        'domain': _top_counts(domains),
+        'intent': _top_counts(intents),
+        'func': _top_counts(funcs),
+        'top_queries': _top_counts(top_queries, limit=20),
+    }
+
+
+def _stride_sample(items: list[dict[str, Any]], sample_size: int) -> list[dict[str, Any]]:
+    if sample_size >= len(items):
+        return list(items)
+    if sample_size == 1:
+        return [items[0]]
+    step = (len(items) - 1) / (sample_size - 1)
+    return [items[round(index * step)] for index in range(sample_size)]
+
+
+def _session_example(session: dict[str, Any]) -> dict[str, Any]:
+    return {
+        'semantic_session_id': session.get('semantic_session_id'),
+        'req_id': session.get('req_id'),
+        'device': session.get('device'),
+        'turn_count': session.get('turn_count'),
+        'date': session.get('date'),
+        'turns': [
+            {
+                'timestamp': turn.get('timestamp'),
+                'query': turn.get('query'),
+                'tts': turn.get('tts'),
+                'action': turn.get('action'),
+            }
+            for turn in session.get('turns', [])[:5]
+        ],
+    }
+
+
+def _top_counts(counter: Counter[str], *, limit: int = 20) -> list[dict[str, Any]]:
+    return [{'value': value, 'count': count} for value, count in counter.most_common(limit)]
+
+
+def _counter_key(value: Any) -> str:
+    if value is None or value == '':
+        return '<empty>'
+    return str(value)

src/skills/bundled/online-badcase-mining/SKILL.md

@@ -3,11 +3,43 @@ name: online-badcase-mining
 description: 分析 badcase，并通过可 review 的策略迭代挖掘线上相似 case。
 when_to_use: 当用户提供线上 badcase 或产品/标签定义，并希望查找相似线上问题或构建专项集时使用。
 aliases: badcase-mining, online-mining
-allowed_tools: read_file, write_file, edit_file, grep_search, glob_search, ask_user_question, mcp_list_tools, mcp_call_tool
+allowed_tools: read_file, write_file, edit_file, grep_search, glob_search, ask_user_question, data_agent_profile_router_sessions, data_agent_search_router_sessions, data_agent_sample_router_candidates, data_agent_convert_router_candidates_to_records, data_agent_export_dataset_records
 ---
 
 使用这个 skill 处理“badcase 或标签定义 -> 挖掘策略 -> 候选召回 -> 抽样 review -> 策略迭代 -> mined dataset”的工作流。
 
+## 两条分支
+
+线上挖掘后必须先判断用户要的是哪条分支。
+
+### 分支 A：线上候选直接作为数据样本
+
+当用户说“把筛选出的数据变成样本”“拿这些线上数据做评测集”“保留这批线上 case”“导出候选样本”时，走这个分支。
+
+流程：
+
+1. 用 `data_agent_profile_router_sessions` 查看数据概貌。
+2. 用 `data_agent_search_router_sessions` 按策略召回线上候选。
+3. 用 `data_agent_sample_router_candidates` 抽样展示给用户 review。
+4. 根据用户 review 意见形成 include/exclude/uncertain 决策和目标标签。
+5. 用 `data_agent_convert_router_candidates_to_records` 把线上候选直接转换为 canonical records。
+6. 如果用户要求落盘，用 `data_agent_export_dataset_records` 导出紧凑 JSONL。
+
+这个分支不生成新 query，不调用数据生成计划工具，不调用 dataset draft 归一化工具。
+
+### 分支 B：基于线上问题再生成补充数据
+
+当用户明确说“生成/扩写/构造/造一批类似 case/补充训练数据”时，才走这个分支。
+
+流程：
+
+1. 先完成线上召回、抽样和 review。
+2. 总结线上问题模式和需要覆盖的边界。
+3. 再切换到数据生成流程，创建待 review 的 generation goal 和 generation plan。
+4. 用户确认后才生成 draft，并转换为 canonical records。
+
+如果用户只是要求“把线上数据作为样本”，不要进入这个分支。
+
 ## 输入假设
 
 用户可能会提供：
@@ -51,15 +83,16 @@ allowed_tools: read_file, write_file, edit_file, grep_search, glob_search, ask_u
 - `grep_search`
 - `glob_search`
 - `ask_user_question`
+- `data_agent_profile_router_sessions`：读取 `router_session_parquet/date=YYYYMMDD/` 的小样本概貌，查看 schema、设备、轮次、domain、intent、func 等分布。
+- `data_agent_search_router_sessions`：按日期、设备、domain、intent、func、query 关键词/正则、轮次数等条件召回线上 session 候选；输出命中 turn、前文 turn、req_id 和 action_json 解析结果。
+- `data_agent_sample_router_candidates`：对召回候选做稳定抽样，并输出 review 需要的基础统计。
+- `data_agent_convert_router_candidates_to_records`：把 review 后的线上候选直接转换为 canonical records；用于“线上数据作为样本”的分支。
+- `data_agent_export_dataset_records`：后续把 review 后的线上候选转换为 canonical records 后落盘；默认紧凑 JSONL。
 
-如果线上挖掘工具通过 MCP 暴露，先用 `mcp_list_tools` 查看可用工具，再决定是否调用 `mcp_call_tool`。
+TODO：后续规划中的专用工具：
-
-TODO：规划中的专用工具：
 
 - `analyze_badcases`
 - `build_mining_strategy`
-- `search_online_sessions`
-- `sample_online_candidates`
 - `create_annotation_batch`
 - `read_annotation_result`
 - `evaluate_mining_precision`
@@ -86,3 +119,5 @@ tasks/{task_id}/memory/failed_attempts.md
 - 除非批准的工具输出已经脱敏，否则不要导出敏感线上原始字段。
 - 不要把第一版挖掘策略当成最终策略。
 - 始终让筛选条件和抽样决策可 review。
+- 只有用户明确要求生成、扩写或构造新数据时，才进入数据生成分支。
+- 如果用户要求把筛选出的线上候选变成样本，必须先使用 `data_agent_convert_router_candidates_to_records`，不要改走 generation goal/plan。

tests/test_data_agent_router_sessions.py

@@ -0,0 +1,229 @@
+from __future__ import annotations
+
+import importlib.util
+import json
+import tempfile
+import unittest
+from pathlib import Path
+
+from src.agent_tools import build_tool_context, default_tool_registry, execute_tool
+from src.agent_types import AgentRuntimeConfig
+from src.data_agent_router_sessions import (
+    convert_router_candidates_to_records,
+    profile_router_sessions,
+    sample_router_candidates,
+    search_router_sessions,
+)
+
+
+HAS_PYARROW = importlib.util.find_spec('pyarrow') is not None
+
+
+class DataAgentRouterSessionTests(unittest.TestCase):
+    def test_sample_router_candidates_accepts_search_payload(self) -> None:
+        payload = {
+            'candidates': [
+                _candidate('s1', 'phone', 'QA', 'unknown', '怎么查找设备'),
+                _candidate('s2', 'car', 'mapCopilot', '导航', '导航去公司'),
+                _candidate('s3', 'phone', 'QA', 'unknown', '手机丢了怎么办'),
+            ]
+        }
+
+        result = sample_router_candidates(payload, sample_size=2, strategy='stride')
+
+        self.assertEqual(result['total_candidate_count'], 3)
+        self.assertEqual(result['sampled_count'], 2)
+        self.assertEqual(result['summary']['device'][0], {'value': 'phone', 'count': 2})
+        self.assertEqual(result['candidates'][0]['semantic_session_id'], 's1')
+        self.assertEqual(result['candidates'][1]['semantic_session_id'], 's3')
+
+    def test_sample_router_candidates_accepts_json_string(self) -> None:
+        result = sample_router_candidates(
+            json.dumps([_candidate('s1', 'phone', 'QA', 'unknown', '怎么查找设备')], ensure_ascii=False),
+            sample_size=1,
+        )
+
+        self.assertEqual(result['sampled_count'], 1)
+        self.assertEqual(result['candidates'][0]['matched_turn']['query'], '怎么查找设备')
+
+    def test_convert_router_candidates_to_records_keeps_online_metadata(self) -> None:
+        candidate = _candidate('s1', 'phone', 'QA', 'unknown', '总结一下')
+        candidate['turn_count'] = 2
+        candidate['matched_turn_index'] = 1
+        candidate['matched_turn']['timestamp'] = 20
+        candidate['prev_turns'] = [{'timestamp': 10, 'query': '这篇文章讲了什么', 'tts': ''}]
+
+        result = convert_router_candidates_to_records(
+            [candidate],
+            dataset_label='总结类边界评测集',
+            default_target='Summarize',
+            batch_id='summary',
+        )
+
+        self.assertEqual(result['record_count'], 1)
+        record = result['records'][0]
+        self.assertEqual(record['source']['type'], 'online')
+        self.assertEqual(record['source']['request_id'], 's1-rid')
+        self.assertEqual(record['turn']['query'], '总结一下')
+        self.assertEqual(record['prev_session'], [{'query': '这篇文章讲了什么', 'tts': '', 'timestamp': 10}])
+        self.assertEqual(record['context']['domain'], 'QA')
+        self.assertEqual(record['label']['target'], 'Summarize')
+
+    def test_convert_router_candidates_to_records_applies_review_decisions(self) -> None:
+        candidates = [
+            _candidate('s1', 'phone', 'QA', 'unknown', '总结一下'),
+            _candidate('s2', 'phone', 'QA', 'unknown', '总结一下红楼梦'),
+        ]
+
+        result = convert_router_candidates_to_records(
+            candidates,
+            dataset_label='总结类边界评测集',
+            review_decisions=[
+                {
+                    'semantic_session_id': 's1',
+                    'matched_turn_index': 0,
+                    'decision': 'include',
+                    'target': 'Summarize',
+                    'notes': '前文有可总结内容',
+                },
+                {
+                    'semantic_session_id': 's2',
+                    'matched_turn_index': 0,
+                    'decision': 'exclude',
+                    'notes': '开放知识对象',
+                },
+            ],
+        )
+
+        self.assertEqual(result['record_count'], 1)
+        self.assertEqual(result['skipped_count'], 1)
+        self.assertEqual(result['records'][0]['meta']['notes'], '前文有可总结内容')
+
+    @unittest.skipUnless(HAS_PYARROW, 'pyarrow is required for parquet tests')
+    def test_profile_and_search_router_sessions_read_parquet(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            _write_router_parquet(Path(tmp_dir))
+
+            profile = profile_router_sessions(tmp_dir, dates=['20260428'], max_files=1)
+            search = search_router_sessions(
+                tmp_dir,
+                dates=['20260428'],
+                domains=['mapCopilot'],
+                query_keywords=['导航'],
+                max_files=1,
+                max_candidates=5,
+            )
+
+        self.assertEqual(profile['sampled_row_count'], 2)
+        self.assertEqual(profile['missing_req_id_count'], 0)
+        self.assertTrue(any(item['value'] == 'phone' for item in profile['distributions']['device']))
+        self.assertEqual(search['candidate_count'], 1)
+        candidate = search['candidates'][0]
+        self.assertEqual(candidate['req_id'], 'rid-1')
+        self.assertEqual(candidate['matched_turn']['query'], '导航去公司')
+        self.assertEqual(candidate['prev_turns'][0]['query'], '你好小爱')
+
+    @unittest.skipUnless(HAS_PYARROW, 'pyarrow is required for parquet tests')
+    def test_router_session_tools_execute_against_registry(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            _write_router_parquet(Path(tmp_dir))
+            context = build_tool_context(AgentRuntimeConfig(cwd=Path(tmp_dir)))
+            registry = default_tool_registry()
+
+            search_result = execute_tool(
+                registry,
+                'data_agent_search_router_sessions',
+                {'dates': ['20260428'], 'query_regex': '查找设备|手机丢了', 'max_files': 1},
+                context,
+            )
+            self.assertTrue(search_result.ok, search_result.content)
+            sample_result = execute_tool(
+                registry,
+                'data_agent_sample_router_candidates',
+                {'candidates': json.loads(search_result.content), 'sample_size': 1, 'strategy': 'first'},
+                context,
+            )
+            convert_result = execute_tool(
+                registry,
+                'data_agent_convert_router_candidates_to_records',
+                {
+                    'candidates': json.loads(sample_result.content),
+                    'dataset_label': '查找设备评测集',
+                    'default_target': 'QA',
+                    'batch_id': 'find_device',
+                },
+                context,
+            )
+
+        self.assertTrue(sample_result.ok, sample_result.content)
+        self.assertEqual(json.loads(sample_result.content)['sampled_count'], 1)
+        self.assertTrue(convert_result.ok, convert_result.content)
+        self.assertTrue(json.loads(convert_result.content)['validation']['ok'])
+
+
+def _candidate(session_id: str, device: str, domain: str, intent: str, query: str) -> dict[str, object]:
+    return {
+        'semantic_session_id': session_id,
+        'req_id': f'{session_id}-rid',
+        'device': device,
+        'turn_count': 1,
+        'matched_turn_index': 0,
+        'matched_turn': {
+            'timestamp': 1,
+            'query': query,
+            'tts': '',
+            'action': {'domain': domain, 'intent': intent},
+        },
+        'prev_turns': [],
+        'source_path': 'router_session_parquet/date=20260428/part-0.parquet',
+        'date': '20260428',
+    }
+
+
+def _write_router_parquet(root: Path) -> None:
+    import pyarrow as pa
+    import pyarrow.parquet as pq
+
+    partition = root / 'router_session_parquet' / 'date=20260428'
+    partition.mkdir(parents=True)
+    rows = [
+        {
+            'semantic_session_id': 'phone_20260428_1',
+            'device': 'phone',
+            'req_id': 'rid-1',
+            'turn_count': 2,
+            'turns_array': [
+                {
+                    'timestamp': 1,
+                    'query': '你好小爱',
+                    'action_json': '{"domain":"dialogCopilot","intent":"问候"}',
+                    'tts': '你好',
+                },
+                {
+                    'timestamp': 2,
+                    'query': '导航去公司',
+                    'action_json': '{"domain":"mapCopilot","intent":"导航","func":"start_navigation"}',
+                    'tts': '开始导航',
+                },
+            ],
+        },
+        {
+            'semantic_session_id': 'phone_20260428_2',
+            'device': 'phone',
+            'req_id': 'rid-2',
+            'turn_count': 1,
+            'turns_array': [
+                {
+                    'timestamp': 3,
+                    'query': '手机丢了怎么查找设备',
+                    'action_json': '{"domain":"QA","intent":"unknown"}',
+                    'tts': '',
+                }
+            ],
+        },
+    ]
+    pq.write_table(pa.Table.from_pylist(rows), partition / 'part-0.parquet')
+
+
+if __name__ == '__main__':
+    unittest.main()