from __future__ import annotations

"""数据智能体记录辅助逻辑。

这个模块把数据智能体专属的记录处理逻辑从通用工具注册表里拆出来。
Skill 应该先让模型生成紧凑、便于人工 review 的 dataset draft text，
再由这里的逻辑解析并补齐为 canonical records，供后续校验和导出使用。
"""

import csv
import json
import re
import time
from io import StringIO
from pathlib import Path
from typing import Any


DEFAULT_REQUEST_ID = 'aabbccdd'
DEFAULT_TIMESTAMP_STEP_MS = 60_000
DEFAULT_SYSTEM_PROMPT = '你是小爱同学，中文智能语音助手。'
DEFAULT_CONTEXT_FIELDS = ('location', 'rag')
DEFAULT_PLAN_STATE_FILE = '.port_sessions/data_agent_generation_plans.json'
DEFAULT_GOAL_STATE_FILE = '.port_sessions/data_agent_generation_goals.json'


class DataRecordError(ValueError):
    """Raised when dataset draft text or records are structurally invalid."""


def prepare_generation_goal(
    *,
    root: str,
    dataset_label: str,
    goal_summary: str,
    coverage: str,
    exclusions: str,
    target: str = '',
    target_definitions: list[dict[str, Any]] | None = None,
    plan_hint: str = '',
    open_questions: list[str] | None = None,
    source_refs: list[str] | None = None,
    notes: str = '',
) -> dict[str, Any]:
    missing = _missing_plan_fields(
        {
            'dataset_label': dataset_label,
            'goal_summary': goal_summary,
            'coverage': coverage,
            'exclusions': exclusions,
        }
    )
    if missing:
        raise DataRecordError('missing required goal fields: ' + ', '.join(missing))
    normalized_targets = _normalize_target_definitions(target_definitions)
    target = _normalize_target_expression(target)
    if not target.strip() and not normalized_targets:
        raise DataRecordError('generation goal must include target or target_definitions')
    state = _load_goal_state(root)
    sequence = int(state.get('next_sequence') or 1)
    goal_id = f'data_goal_{sequence:06d}'
    goal = {
        'goal_id': goal_id,
        'status': 'pending_confirmation',
        'dataset_label': dataset_label.strip(),
        'goal_summary': goal_summary.strip(),
        'target': target,
        'target_definitions': normalized_targets,
        'plan_hint': plan_hint.strip(),
        'coverage': coverage.strip(),
        'exclusions': exclusions.strip(),
        'open_questions': _normalize_string_list(open_questions, 'open_questions'),
        'source_refs': _normalize_string_list(source_refs, 'source_refs'),
        'notes': notes.strip(),
        'revision': 1,
        'created_at_ms': int(time.time() * 1000),
    }
    goals = state.setdefault('goals', {})
    if not isinstance(goals, dict):
        goals = {}
        state['goals'] = goals
    goals[goal_id] = goal
    state['next_sequence'] = sequence + 1
    _save_goal_state(root, state)
    return {
        'goal': goal,
        'message': 'generation goal 已创建，状态为 pending_confirmation。请向用户展示 goal，并等待用户确认后再创建 generation plan。',
    }


def confirm_generation_goal(
    *,
    root: str,
    goal_id: str,
    confirmation: str,
    reviewed_revision: int | None = None,
) -> dict[str, Any]:
    if not goal_id.strip():
        raise DataRecordError('goal_id is required')
    if not _is_confirmation_text(confirmation):
        raise DataRecordError('confirmation must clearly approve the generation goal')
    state = _load_goal_state(root)
    goal = _get_goal(state, goal_id)
    current_revision = int(goal.get('revision') or 1)
    if reviewed_revision is not None and reviewed_revision != current_revision:
        raise DataRecordError(
            f'reviewed_revision must match current goal revision {current_revision}'
        )
    if not _allowed_targets_for_goal(goal):
        raise DataRecordError('confirmed generation goal must include target or target_definitions')
    goal['status'] = 'confirmed'
    goal['confirmed_at_ms'] = int(time.time() * 1000)
    goal['confirmation'] = confirmation.strip()
    goal['confirmed_revision'] = current_revision
    _save_goal_state(root, state)
    return {
        'goal': goal,
        'confirmed_goal_id': goal_id,
        'message': 'generation goal 已确认，可以把 confirmed_goal_id 传给 data_agent_prepare_generation_plan。',
    }


def normalize_dataset_draft(
    draft_text: str,
    *,
    batch_id: str = DEFAULT_REQUEST_ID,
    source_type: str = 'generated',
    base_timestamp: int | None = None,
    timestamp_step_ms: int = DEFAULT_TIMESTAMP_STEP_MS,
    default_request_id: str = DEFAULT_REQUEST_ID,
    confirmed_plan_id: str | None = None,
    plan_state_root: str | None = None,
) -> dict[str, Any]:
    if not draft_text.strip():
        raise DataRecordError('draft_text must be non-empty')
    if timestamp_step_ms <= 0:
        raise DataRecordError('timestamp_step_ms must be greater than 0')
    confirmed_plan: dict[str, Any] | None = None
    if plan_state_root is not None:
        if not confirmed_plan_id:
            raise DataRecordError('confirmed_plan_id is required before normalizing generated data')
        confirmed_plan = _require_confirmed_generation_plan(plan_state_root, confirmed_plan_id)

    base = int(time.time() * 1000) if base_timestamp is None else base_timestamp
    parsed = _parse_draft_text(draft_text)
    records: list[dict[str, Any]] = []
    warnings: list[str] = []

    for index, case in enumerate(parsed['cases'], start=1):
        record, case_warnings = _case_to_record(
            case,
            global_dataset_label=parsed.get('dataset_label') or '',
            index=index,
            batch_id=batch_id,
            source_type=source_type,
            base_timestamp=base,
            timestamp_step_ms=timestamp_step_ms,
            default_request_id=default_request_id,
        )
        records.append(record)
        warnings.extend(case_warnings)

    if confirmed_plan is not None:
        _validate_records_against_plan(records, confirmed_plan)

    return {'records': records, 'warnings': warnings}


def prepare_generation_plan(
    *,
    root: str,
    dataset_label: str,
    target: str,
    total_count: int,
    turn_mix: str,
    coverage: str,
    exclusions: str,
    output_path: str,
    target_definitions: list[dict[str, Any]] | None = None,
    notes: str = '',
    confirmed_goal_id: str | None = None,
) -> dict[str, Any]:
    confirmed_goal: dict[str, Any] | None = None
    if confirmed_goal_id is not None:
        confirmed_goal = _require_confirmed_generation_goal(root, confirmed_goal_id)
    missing = _missing_plan_fields(
        {
            'dataset_label': dataset_label,
            'turn_mix': turn_mix,
            'coverage': coverage,
            'exclusions': exclusions,
            'output_path': output_path,
        }
    )
    target = _normalize_target_expression(target)
    normalized_targets = _normalize_target_definitions(target_definitions)
    if confirmed_goal is not None and not target.strip() and not normalized_targets:
        normalized_targets = _normalize_target_definitions(
            confirmed_goal.get('target_definitions') if isinstance(confirmed_goal, dict) else None
        )
        if not normalized_targets:
            target = _normalize_target_expression(str(confirmed_goal.get('target') or ''))
    if not target.strip() and not normalized_targets:
        missing.append('target')
    if missing:
        raise DataRecordError('missing required plan fields: ' + ', '.join(missing))
    if total_count <= 0:
        raise DataRecordError('total_count must be greater than 0')
    if confirmed_goal is not None:
        _validate_plan_against_goal(
            dataset_label=dataset_label,
            target=target,
            target_definitions=normalized_targets,
            goal=confirmed_goal,
        )

    state = _load_plan_state(root)
    sequence = int(state.get('next_sequence') or 1)
    plan_id = f'data_plan_{sequence:06d}'
    resolved_output_path = _dedupe_output_path(root, output_path.strip(), plan_id)
    plan = {
        'plan_id': plan_id,
        'status': 'pending_confirmation',
        'dataset_label': dataset_label.strip(),
        'target': target,
        'target_definitions': normalized_targets,
        'total_count': total_count,
        'turn_mix': turn_mix.strip(),
        'coverage': coverage.strip(),
        'exclusions': exclusions.strip(),
        'output_path': resolved_output_path,
        'requested_output_path': output_path.strip(),
        'notes': notes.strip(),
        'confirmed_goal_id': confirmed_goal_id or '',
        'revision': 1,
        'review_history': [],
        'created_at_ms': int(time.time() * 1000),
    }
    plans = state.setdefault('plans', {})
    if not isinstance(plans, dict):
        plans = {}
        state['plans'] = plans
    plans[plan_id] = plan
    state['next_sequence'] = sequence + 1
    _save_plan_state(root, state)
    return {
        'plan': plan,
        'message': '计划已创建，状态为 pending_confirmation。请向用户展示计划，并等待用户确认后再调用确认工具。',
    }


def get_generation_plan(*, root: str, plan_id: str) -> dict[str, Any]:
    state = _load_plan_state(root)
    plan = _get_plan(state, plan_id)
    return {
        'plan': plan,
        'message': '请把计划展示给用户 review。用户有意见时调用 update；用户明确确认后再调用 confirm。',
    }


def update_generation_plan(
    *,
    root: str,
    plan_id: str,
    review_feedback: str,
    updates: dict[str, Any],
) -> dict[str, Any]:
    if not review_feedback.strip():
        raise DataRecordError('review_feedback is required')
    if not isinstance(updates, dict) or not updates:
        raise DataRecordError('updates must be a non-empty object')
    state = _load_plan_state(root)
    plan = _get_plan(state, plan_id)
    if plan.get('status') == 'confirmed':
        raise DataRecordError(f'generation plan is already confirmed: {plan_id}')

    allowed_keys = {
        'dataset_label',
        'target',
        'target_definitions',
        'total_count',
        'turn_mix',
        'coverage',
        'exclusions',
        'output_path',
        'notes',
    }
    unknown = sorted(key for key in updates if key not in allowed_keys)
    if unknown:
        raise DataRecordError('unsupported update fields: ' + ', '.join(unknown))

    if 'target_definitions' in updates:
        updates = dict(updates)
        updates['target_definitions'] = _normalize_target_definitions(updates['target_definitions'])
    if 'total_count' in updates:
        value = updates['total_count']
        if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
            raise DataRecordError('total_count must be a positive integer')
    for key, value in updates.items():
        if key == 'target_definitions' or key == 'total_count':
            continue
        if not isinstance(value, str):
            raise DataRecordError(f'{key} must be a string')
        updates[key] = value.strip()

    plan.update(updates)
    plan['status'] = 'pending_confirmation'
    plan['revision'] = int(plan.get('revision') or 1) + 1
    history = plan.get('review_history')
    if not isinstance(history, list):
        history = []
        plan['review_history'] = history
    history.append(
        {
            'revision': plan['revision'],
            'feedback': review_feedback.strip(),
            'updates': updates,
            'updated_at_ms': int(time.time() * 1000),
        }
    )
    _save_plan_state(root, state)
    return {
        'plan': plan,
        'message': '计划已根据用户反馈更新。请重新展示给用户 review，确认前不要生成数据。',
    }


def confirm_generation_plan(
    *,
    root: str,
    plan_id: str,
    confirmation: str,
    reviewed_revision: int | None = None,
) -> dict[str, Any]:
    if not plan_id.strip():
        raise DataRecordError('plan_id is required')
    if not _is_confirmation_text(confirmation):
        raise DataRecordError('confirmation must clearly approve the plan')
    state = _load_plan_state(root)
    plan = _get_plan(state, plan_id)
    current_revision = int(plan.get('revision') or 1)
    if reviewed_revision is not None and reviewed_revision != current_revision:
        raise DataRecordError(
            f'reviewed_revision must match current plan revision {current_revision}'
        )
    plan['status'] = 'confirmed'
    plan['confirmed_at_ms'] = int(time.time() * 1000)
    plan['confirmation'] = confirmation.strip()
    plan['confirmed_revision'] = current_revision
    _save_plan_state(root, state)
    return {
        'plan': plan,
        'confirmed_plan_id': plan_id,
        'message': '计划已确认，可以把 confirmed_plan_id 传给 product-data 的 normalize_dataset_draft.py 脚本。',
    }


def validate_dataset_records(records: list[dict[str, Any]]) -> dict[str, Any]:
    errors: list[dict[str, Any]] = []
    warnings: list[dict[str, Any]] = []

    if not isinstance(records, list):
        raise DataRecordError('records must be a list')

    seen_ids: set[str] = set()
    for index, record in enumerate(records):
        prefix = f'records[{index}]'
        if not isinstance(record, dict):
            errors.append(_issue(prefix, 'record must be an object'))
            continue

        record_id = record.get('record_id')
        if not _non_empty_str(record_id):
            errors.append(_issue(f'{prefix}.record_id', 'record_id is required'))
        elif record_id in seen_ids:
            errors.append(_issue(f'{prefix}.record_id', 'record_id must be unique'))
        else:
            seen_ids.add(record_id)

        source = record.get('source')
        if not isinstance(source, dict):
            errors.append(_issue(f'{prefix}.source', 'source is required'))
        else:
            if source.get('type') not in {'generated', 'online', 'manual', 'mixed'}:
                errors.append(_issue(f'{prefix}.source.type', 'source.type is invalid'))
            if not _non_empty_str(source.get('request_id')):
                errors.append(_issue(f'{prefix}.source.request_id', 'source.request_id is required'))
            if not _int_like(source.get('timestamp')):
                errors.append(_issue(f'{prefix}.source.timestamp', 'source.timestamp must be an integer'))

        turn = record.get('turn')
        if not isinstance(turn, dict):
            errors.append(_issue(f'{prefix}.turn', 'turn is required'))
            turn_timestamp = None
        else:
            if not _non_empty_str(turn.get('query')):
                errors.append(_issue(f'{prefix}.turn.query', 'turn.query is required'))
            if 'tts' in turn:
                errors.append(_issue(f'{prefix}.turn.tts', 'current turn must not contain tts'))
            turn_timestamp = turn.get('timestamp')
            if not _int_like(turn_timestamp):
                errors.append(_issue(f'{prefix}.turn.timestamp', 'turn.timestamp must be an integer'))

        prev_session = record.get('prev_session')
        if not isinstance(prev_session, list):
            errors.append(_issue(f'{prefix}.prev_session', 'prev_session must be a list'))
            prev_timestamps: list[int] = []
        else:
            if len(prev_session) > 10:
                errors.append(_issue(f'{prefix}.prev_session', 'prev_session must contain at most 10 turns'))
            prev_timestamps = []
            for prev_index, item in enumerate(prev_session):
                item_prefix = f'{prefix}.prev_session[{prev_index}]'
                if not isinstance(item, dict):
                    errors.append(_issue(item_prefix, 'prev_session item must be an object'))
                    continue
                if not _non_empty_str(item.get('query')):
                    errors.append(_issue(f'{item_prefix}.query', 'query is required'))
                if 'tts' not in item or not isinstance(item.get('tts'), str):
                    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'))
                else:
                    prev_timestamps.append(int(timestamp))

        if prev_timestamps != sorted(prev_timestamps):
            errors.append(_issue(f'{prefix}.prev_session', 'timestamps must be sorted from early to late'))
        if _int_like(turn_timestamp):
            all_timestamps = [*prev_timestamps, int(turn_timestamp)]
            for left, right in zip(all_timestamps, all_timestamps[1:]):
                if right <= left:
                    errors.append(_issue(f'{prefix}.timestamp', 'turn timestamps must be strictly increasing'))
                    break
                if right - left > 300_000:
                    warnings.append(_issue(f'{prefix}.timestamp', 'adjacent turns are more than 5 minutes apart'))

        context = record.get('context')
        if not isinstance(context, dict):
            errors.append(_issue(f'{prefix}.context', 'context must be an object'))

        label = record.get('label')
        if not isinstance(label, dict):
            errors.append(_issue(f'{prefix}.label', 'label is required'))
        else:
            if not _non_empty_str(label.get('dataset_label')):
                warnings.append(_issue(f'{prefix}.label.dataset_label', 'dataset_label is empty'))
            if not _non_empty_str(label.get('target')):
                errors.append(_issue(f'{prefix}.label.target', 'label.target is required'))
            if label.get('target_type') not in {'agent', 'function', 'unknown'}:
                errors.append(_issue(f'{prefix}.label.target_type', 'label.target_type is invalid'))

        dimensions = record.get('dimensions')
        if not isinstance(dimensions, dict):
            warnings.append(_issue(f'{prefix}.dimensions', 'dimensions.complex is missing; false will be used as fallback'))
        elif 'complex' not in dimensions:
            warnings.append(_issue(f'{prefix}.dimensions.complex', 'complex is missing; false will be used as fallback'))
        elif not isinstance(dimensions.get('complex'), bool):
            errors.append(_issue(f'{prefix}.dimensions.complex', 'complex must be a boolean'))

    return {
        'ok': not errors,
        'error_count': len(errors),
        'warning_count': len(warnings),
        'errors': errors,
        'warnings': warnings,
    }


def export_dataset_records(
    records: list[dict[str, Any]],
    *,
    root: str,
    output_path: str,
    output_format: str = 'jsonl',
    require_validation_ok: bool = True,
    overwrite: bool = True,
    export_table: bool = True,
) -> dict[str, Any]:
    """校验并导出 canonical records。

    这个函数用于收口数据智能体的最终落盘动作，避免模型手写 JSON/JSONL。
    默认导出紧凑 JSONL：一行一个 canonical record，方便 diff、抽样和增量处理。
    """

    if output_format not in {'jsonl', 'json'}:
        raise DataRecordError('output_format must be jsonl or json')
    if not output_path.strip():
        raise DataRecordError('output_path is required')

    validation = validate_dataset_records(records)
    if require_validation_ok and not validation['ok']:
        raise DataRecordError(
            f'records failed validation with {validation["error_count"]} errors'
        )

    path = _resolve_output_path(root, output_path)
    table_path = path.with_name('records.csv')
    if path.exists() and not overwrite:
        raise DataRecordError(f'output_path already exists: {output_path}')
    if export_table and table_path.exists() and not overwrite:
        raise DataRecordError('table output_path already exists: records.csv')
    path.parent.mkdir(parents=True, exist_ok=True)

    if output_format == 'jsonl':
        content = ''.join(
            json.dumps(record, ensure_ascii=False, separators=(',', ':')) + '\n'
            for record in records
        )
    else:
        content = json.dumps(records, ensure_ascii=False, separators=(',', ':')) + '\n'

    path.write_text(content, encoding='utf-8')
    rel_path = str(path.relative_to(Path(root).resolve()))
    payload = {
        'output_path': rel_path,
        'output_format': output_format,
        'record_count': len(records),
        'bytes_written': len(content.encode('utf-8')),
        'validation': validation,
    }
    if export_table:
        table_content = render_dataset_records_table_csv(records)
        table_path.write_text(table_content, encoding='utf-8-sig')
        payload.update(
            {
                'table_output_path': str(table_path.relative_to(Path(root).resolve())),
                'table_output_format': 'csv',
                'table_bytes_written': len(table_content.encode('utf-8-sig')),
            }
        )
    return payload


def export_training_jsonl(
    records: list[dict[str, Any]],
    *,
    root: str,
    output_path: str,
    session_num: int = 5,
    session_time_minutes: int = 5,
    context_fields: list[str] | None = None,
    system_prompt: str = DEFAULT_SYSTEM_PROMPT,
    require_validation_ok: bool = True,
    overwrite: bool = True,
) -> dict[str, Any]:
    """导出训练集 JSONL：system / instruction / output。"""

    if session_num <= 0:
        raise DataRecordError('session_num must be greater than 0')
    if session_time_minutes <= 0:
        raise DataRecordError('session_time_minutes must be greater than 0')
    validation = validate_dataset_records(records)
    if require_validation_ok and not validation['ok']:
        raise DataRecordError(
            f'records failed validation with {validation["error_count"]} errors'
        )
    path = _resolve_output_path(root, output_path)
    if path.exists() and not overwrite:
        raise DataRecordError(f'output_path already exists: {output_path}')
    path.parent.mkdir(parents=True, exist_ok=True)
    fields = _normalize_context_fields(context_fields)
    content = ''.join(
        _training_jsonl_line(
            record,
            session_num=session_num,
            session_time_minutes=session_time_minutes,
            context_fields=fields,
            system_prompt=system_prompt,
        )
        + '\n'
        for record in records
    )
    path.write_text(content, encoding='utf-8')
    return {
        'output_path': str(path.relative_to(Path(root).resolve())),
        'output_format': 'jsonl',
        'record_count': len(records),
        'bytes_written': len(content.encode('utf-8')),
        'validation': validation,
    }


def export_planning_eval_csv(
    records: list[dict[str, Any]],
    *,
    root: str,
    output_path: str,
    session_num: int = 5,
    session_time_minutes: int = 5,
    context_fields: list[str] | None = None,
    system_prompt: str = DEFAULT_SYSTEM_PROMPT,
    complex_default: bool = False,
    require_validation_ok: bool = True,
    overwrite: bool = True,
) -> dict[str, Any]:
    """导出评测 CSV：request_id / newPrompt / query / 类别真实标签 / code标签 / complex。"""

    if session_num <= 0:
        raise DataRecordError('session_num must be greater than 0')
    if session_time_minutes <= 0:
        raise DataRecordError('session_time_minutes must be greater than 0')
    validation = validate_dataset_records(records)
    if require_validation_ok and not validation['ok']:
        raise DataRecordError(
            f'records failed validation with {validation["error_count"]} errors'
        )
    path = _resolve_output_path(root, output_path)
    if path.exists() and not overwrite:
        raise DataRecordError(f'output_path already exists: {output_path}')
    path.parent.mkdir(parents=True, exist_ok=True)
    fields = _normalize_context_fields(context_fields)
    content = render_planning_eval_csv(
        records,
        session_num=session_num,
        session_time_minutes=session_time_minutes,
        context_fields=fields,
        system_prompt=system_prompt,
        complex_default=complex_default,
    )
    path.write_text(content, encoding='utf-8-sig')
    return {
        'output_path': str(path.relative_to(Path(root).resolve())),
        'output_format': 'csv',
        'record_count': len(records),
        'bytes_written': len(content.encode('utf-8-sig')),
        'validation': validation,
    }


def render_dataset_records_table_csv(records: list[dict[str, Any]]) -> str:
    """把 canonical records 转成同事流转用的表格 CSV。"""

    output = StringIO()
    writer = csv.DictWriter(output, fieldnames=_dataset_table_columns(), lineterminator='\n')
    writer.writeheader()
    for record in records:
        writer.writerow(_dataset_record_table_row(record))
    return output.getvalue()


def render_planning_eval_csv(
    records: list[dict[str, Any]],
    *,
    session_num: int = 5,
    session_time_minutes: int = 5,
    context_fields: list[str] | None = None,
    system_prompt: str = DEFAULT_SYSTEM_PROMPT,
    complex_default: bool = False,
) -> str:
    """把 canonical records 转成含 newPrompt 的评测 CSV。"""

    fields = _normalize_context_fields(context_fields)
    output = StringIO()
    writer = csv.DictWriter(
        output,
        fieldnames=['request_id', 'newPrompt', 'query', '类别真实标签', 'code标签', 'complex'],
        lineterminator='\n',
    )
    writer.writeheader()
    for record in records:
        source = record.get('source') if isinstance(record.get('source'), dict) else {}
        turn = record.get('turn') if isinstance(record.get('turn'), dict) else {}
        target = _record_target(record)
        writer.writerow(
            {
                'request_id': str(source.get('request_id') or ''),
                'newPrompt': build_planning_prompt(
                    record,
                    session_num=session_num,
                    session_time_minutes=session_time_minutes,
                    context_fields=fields,
                    system_prompt=system_prompt,
                ),
                'query': str(turn.get('query') or ''),
                '类别真实标签': _category_label_from_target(target),
                'code标签': target,
                'complex': _eval_complex_literal(_record_complex(record, default=complex_default)),
            }
        )
    return output.getvalue()


def build_training_instruction(
    record: dict[str, Any],
    *,
    session_num: int = 5,
    session_time_minutes: int = 5,
    context_fields: list[str] | None = None,
) -> str:
    turn = record.get('turn') if isinstance(record.get('turn'), dict) else {}
    query = str(turn.get('query') or '')
    context = record.get('context') if isinstance(record.get('context'), dict) else {}
    prev_session = record.get('prev_session') if isinstance(record.get('prev_session'), list) else []
    current_ts = _optional_int_for_export(turn.get('timestamp'))
    if current_ts is None:
        source = record.get('source') if isinstance(record.get('source'), dict) else {}
        current_ts = _optional_int_for_export(source.get('timestamp'))
    session, last_tts = _training_history(
        prev_session,
        current_ts=current_ts,
        session_num=session_num,
        session_time_minutes=session_time_minutes,
    )
    fields = _normalize_context_fields(context_fields)

    instruction = '请参考用户的[当前query]、[对话历史]、[知识注入]、[系统状态]识别出[当前query]的[function]结果，[function]是python的code形式。\n'
    instruction += '[知识注入]\n'
    instruction += f'{_context_prompt_block(context, fields)}\n'
    instruction += '[系统状态]\n'
    instruction += '{}\n'
    instruction += '[对话历史]\n'
    if session:
        history_parts = [f'用户: {session_query}' for session_query in session]
        if last_tts is not None:
            history_parts.append(f'小爱: {last_tts}')
        instruction += '\n'.join(history_parts)
        instruction += '\n'
    instruction += '[当前query]\n'
    instruction += f'用户: {query}\n'
    instruction += '[function]\n'
    return instruction


def build_planning_prompt(
    record: dict[str, Any],
    *,
    session_num: int = 5,
    session_time_minutes: int = 5,
    context_fields: list[str] | None = None,
    system_prompt: str = DEFAULT_SYSTEM_PROMPT,
) -> str:
    """把训练 instruction 包成评测侧使用的 chat template。"""

    instruction = build_training_instruction(
        record,
        session_num=session_num,
        session_time_minutes=session_time_minutes,
        context_fields=context_fields,
    )
    return (
        f'<|im_start|>system\n{system_prompt}<|im_end|>\n'
        f'<|im_start|>user\n{instruction}<|im_end|>\n'
        '<|im_start|>assistant\n'
    )


def _training_jsonl_line(
    record: dict[str, Any],
    *,
    session_num: int,
    session_time_minutes: int,
    context_fields: list[str],
    system_prompt: str,
) -> str:
    payload = {
        'system': system_prompt,
        'instruction': build_training_instruction(
            record,
            session_num=session_num,
            session_time_minutes=session_time_minutes,
            context_fields=context_fields,
        ),
        'output': _combined_function_label(record),
    }
    return json.dumps(payload, ensure_ascii=False, separators=(',', ':'))


def _training_history(
    prev_session: list[Any],
    *,
    current_ts: int | None,
    session_num: int,
    session_time_minutes: int,
) -> tuple[list[str], str | None]:
    if current_ts is None:
        return [], None
    session_time_ms = session_time_minutes * 60 * 1000
    valid_items: list[tuple[int, str, str]] = []
    for item in prev_session:
        if not isinstance(item, dict):
            continue
        ts = _optional_int_for_export(item.get('timestamp'))
        query = str(item.get('query') or '')
        tts = str(item.get('tts') or '')
        if ts is not None and ts > 0 and query:
            valid_items.append((ts, query, tts))
    valid_items.sort(key=lambda item: item[0])

    session: list[str] = []
    last_tts: str | None = None
    last_ts = current_ts
    count = 0
    for ts, query, tts in reversed(valid_items):
        if last_ts - ts <= session_time_ms and last_ts >= ts:
            session.append(query)
            if count == 0:
                last_tts = tts
            last_ts = ts
            count += 1
            if count >= session_num:
                break
        else:
            break
    session.reverse()
    return session, last_tts


def _context_prompt_block(context: dict[str, Any], fields: list[str]) -> str:
    lines = ['{']
    for index, field in enumerate(fields):
        comma = ',' if index < len(fields) - 1 else ''
        value = str(context.get(field, ''))
        lines.append(f'"{field}": {json.dumps(value, ensure_ascii=False)}{comma}')
    lines.append('}')
    return '\n'.join(lines)


def _normalize_context_fields(context_fields: list[str] | None) -> list[str]:
    fields = context_fields or list(DEFAULT_CONTEXT_FIELDS)
    normalized = [field.strip() for field in fields if isinstance(field, str) and field.strip()]
    return normalized or list(DEFAULT_CONTEXT_FIELDS)


def _optional_int_for_export(value: Any) -> int | None:
    if isinstance(value, bool) or value is None or value == '':
        return None
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


def _category_label_from_target(target: str) -> str:
    target = _strip_complex_prefix(target)
    agent_match = re.match(r'''^Agent\s*\(\s*tag\s*=\s*["']([^"']+)["']\s*\)$''', target.strip())
    if agent_match:
        return agent_match.group(1)
    function_names = re.findall(r'\b([A-Za-z_][A-Za-z0-9_]*)\s*\(', target)
    if function_names:
        return function_names[-1]
    return target


def _dataset_table_columns() -> list[str]:
    return [
        'request_id',
        'timestamp',
        'query',
        'prev_session',
        'context',
        'label',
        '是否迁移Function',
        'function',
    ]


def _dataset_record_table_row(record: dict[str, Any]) -> dict[str, str]:
    source = record.get('source') if isinstance(record.get('source'), dict) else {}
    turn = record.get('turn') if isinstance(record.get('turn'), dict) else {}
    label = record.get('label') if isinstance(record.get('label'), dict) else {}
    context = record.get('context') if isinstance(record.get('context'), dict) else {}
    prev_session = record.get('prev_session')
    if not isinstance(prev_session, list):
        prev_session = []
    return {
        'request_id': str(source.get('request_id') or ''),
        'timestamp': str(source.get('timestamp') or turn.get('timestamp') or ''),
        'query': str(turn.get('query') or ''),
        'prev_session': json.dumps(_table_prev_session(prev_session), ensure_ascii=False, separators=(',', ':')),
        'context': json.dumps(context, ensure_ascii=False, separators=(',', ':')),
        'label': str(label.get('dataset_label') or ''),
        '是否迁移Function': '',
        'function': _combined_function_label(record),
    }


def _table_prev_session(prev_session: list[Any]) -> list[dict[str, str]]:
    rows: list[dict[str, str]] = []
    for item in prev_session[-10:]:
        if not isinstance(item, dict):
            continue
        rows.append(
            {
                'query': str(item.get('query') or ''),
                'tts': str(item.get('tts') or ''),
                'timestamp': str(item.get('timestamp') or ''),
            }
        )
    return rows


def _parse_draft_text(draft_text: str) -> dict[str, Any]:
    dataset_label = ''
    cases: list[dict[str, Any]] = []
    current: dict[str, Any] | None = None

    for raw_line in draft_text.splitlines():
        line = raw_line.strip()
        if not line:
            continue
        dataset_match = re.match(r'^#\s*dataset_label\s*[:：]\s*(.+)$', line, re.IGNORECASE)
        if dataset_match:
            dataset_label = dataset_match.group(1).strip()
            continue
        case_match = re.match(r'^#{3,}\s*case\s*[:：]\s*(.+)$', line, re.IGNORECASE)
        if case_match:
            current = {'case_name': case_match.group(1).strip(), 'turns': []}
            cases.append(current)
            continue
        if current is None:
            continue

        field_match = re.match(r'^(用户|小爱|target|complex|notes|dataset_label|request_id|timestamp)\s*[:：]\s*(.*)$', line, re.IGNORECASE)
        if not field_match:
            continue
        key = field_match.group(1).lower()
        value = field_match.group(2).strip()
        if key == '用户':
            current['turns'].append({'role': 'user', 'text': value})
        elif key == '小爱':
            current['turns'].append({'role': 'assistant', 'text': value})
        else:
            current[key] = value

    if not cases:
        raise DataRecordError('draft_text must contain at least one "### case:" block')
    return {'dataset_label': dataset_label, 'cases': cases}


def _missing_plan_fields(fields: dict[str, str]) -> list[str]:
    return [name for name, value in fields.items() if not isinstance(value, str) or not value.strip()]


def _dedupe_output_path(root: str, output_path: str, plan_id: str) -> str:
    from pathlib import Path

    path = Path(output_path)
    # 会话 output 下的标准 records 文件名需要保持稳定；导出工具默认允许覆盖。
    if path.name in {'records.jsonl', 'records.json'} and 'output' in path.parts:
        return output_path
    resolved = path if path.is_absolute() else Path(root).resolve() / path
    if not resolved.exists():
        return output_path
    parts = path.parts
    if len(parts) >= 2 and parts[-1] == 'artifacts':
        task_dir = Path(*parts[:-1])
        deduped = Path(f'{task_dir}-{plan_id}') / 'artifacts'
    else:
        deduped = Path(f'{output_path.rstrip("/")}-{plan_id}')
    return deduped.as_posix()


def _normalize_target_definitions(value: list[dict[str, Any]] | None) -> list[dict[str, str]]:
    if value is None:
        return []
    if not isinstance(value, list):
        raise DataRecordError('target_definitions must be a list')
    normalized: list[dict[str, str]] = []
    for index, item in enumerate(value):
        if not isinstance(item, dict):
            raise DataRecordError(f'target_definitions[{index}] must be an object')
        name = str(item.get('name') or '').strip()
        target = _normalize_target_expression(str(item.get('target') or ''))
        rule = str(item.get('rule') or '').strip()
        if not target:
            raise DataRecordError(f'target_definitions[{index}].target is required')
        normalized.append({'name': name, 'target': target, 'rule': rule})
    return normalized


def _normalize_string_list(value: list[str] | None, field_name: str) -> list[str]:
    if value is None:
        return []
    if not isinstance(value, list):
        raise DataRecordError(f'{field_name} must be a list')
    normalized: list[str] = []
    for index, item in enumerate(value):
        if not isinstance(item, str):
            raise DataRecordError(f'{field_name}[{index}] must be a string')
        text = item.strip()
        if text:
            normalized.append(text)
    return normalized


def _allowed_targets_for_goal(goal: dict[str, Any]) -> set[str]:
    return _allowed_targets_for_plan(goal)


def _allowed_targets_for_plan(plan: dict[str, Any]) -> set[str]:
    target_definitions = plan.get('target_definitions')
    allowed: set[str] = set()
    if isinstance(target_definitions, list):
        for item in target_definitions:
            if isinstance(item, dict) and isinstance(item.get('target'), str) and item['target'].strip():
                allowed.add(_normalize_target_expression(item['target']))
    target = plan.get('target')
    if isinstance(target, str) and target.strip():
        allowed.add(_normalize_target_expression(target))
    return allowed


def _validate_plan_against_goal(
    *,
    dataset_label: str,
    target: str,
    target_definitions: list[dict[str, str]],
    goal: dict[str, Any],
) -> None:
    goal_label = str(goal.get('dataset_label') or '').strip()
    if goal_label and dataset_label.strip() != goal_label:
        raise DataRecordError(
            f'plan dataset_label must match confirmed generation goal: {goal_label}'
        )
    goal_targets = _allowed_targets_for_goal(goal)
    plan_targets = set()
    if target.strip():
        plan_targets.add(_normalize_target_expression(target))
    for item in target_definitions:
        item_target = item.get('target', '').strip()
        if item_target:
            plan_targets.add(item_target)
    unexpected = sorted(plan_targets - goal_targets)
    if unexpected:
        raise DataRecordError(
            'plan targets must match confirmed generation goal: ' + ', '.join(unexpected)
        )


def _validate_records_against_plan(records: list[dict[str, Any]], plan: dict[str, Any]) -> None:
    allowed_targets = _allowed_targets_for_plan(plan)
    if not allowed_targets:
        return
    unexpected: list[str] = []
    for index, record in enumerate(records):
        target = _record_target(record)
        if target and target in allowed_targets:
            continue
        unexpected.append(f'records[{index}].label.target={target!r}')
    if unexpected:
        raise DataRecordError(
            'record targets must match the confirmed generation plan: '
            + '; '.join(unexpected)
        )


def _is_confirmation_text(text: str) -> bool:
    if not isinstance(text, str):
        return False
    normalized = text.strip().lower()
    if not normalized:
        return False
    approval_words = (
        '确认',
        '同意',
        '开始生成',
        '按这个计划',
        '批准',
        'approve',
        'approved',
        'confirmed',
        'yes',
    )
    return any(word in normalized for word in approval_words)


def _plan_state_path(root: str):
    from pathlib import Path

    return Path(root).resolve() / DEFAULT_PLAN_STATE_FILE


def _goal_state_path(root: str):
    from pathlib import Path

    return Path(root).resolve() / DEFAULT_GOAL_STATE_FILE


def _load_plan_state(root: str) -> dict[str, Any]:
    path = _plan_state_path(root)
    if not path.exists():
        return {'next_sequence': 1, 'plans': {}}
    try:
        payload = json.loads(path.read_text(encoding='utf-8'))
    except json.JSONDecodeError as exc:
        raise DataRecordError(f'invalid generation plan state file: {path}') from exc
    if not isinstance(payload, dict):
        raise DataRecordError(f'invalid generation plan state file: {path}')
    if not isinstance(payload.get('plans'), dict):
        payload['plans'] = {}
    if not isinstance(payload.get('next_sequence'), int):
        payload['next_sequence'] = 1
    return payload


def _load_goal_state(root: str) -> dict[str, Any]:
    path = _goal_state_path(root)
    if not path.exists():
        return {'next_sequence': 1, 'goals': {}}
    try:
        payload = json.loads(path.read_text(encoding='utf-8'))
    except json.JSONDecodeError as exc:
        raise DataRecordError(f'invalid generation goal state file: {path}') from exc
    if not isinstance(payload, dict):
        raise DataRecordError(f'invalid generation goal state file: {path}')
    if not isinstance(payload.get('goals'), dict):
        payload['goals'] = {}
    if not isinstance(payload.get('next_sequence'), int):
        payload['next_sequence'] = 1
    return payload


def _save_plan_state(root: str, state: dict[str, Any]) -> None:
    path = _plan_state_path(root)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding='utf-8')


def _save_goal_state(root: str, state: dict[str, Any]) -> None:
    path = _goal_state_path(root)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding='utf-8')


def _get_plan(state: dict[str, Any], plan_id: str) -> dict[str, Any]:
    plans = state.get('plans')
    if not isinstance(plans, dict):
        raise DataRecordError('generation plan state is invalid')
    plan = plans.get(plan_id)
    if not isinstance(plan, dict):
        raise DataRecordError(f'generation plan not found: {plan_id}')
    return plan


def _get_goal(state: dict[str, Any], goal_id: str) -> dict[str, Any]:
    goals = state.get('goals')
    if not isinstance(goals, dict):
        raise DataRecordError('generation goal state is invalid')
    goal = goals.get(goal_id)
    if not isinstance(goal, dict):
        raise DataRecordError(f'generation goal not found: {goal_id}')
    return goal


def _require_confirmed_generation_plan(root: str, plan_id: str) -> dict[str, Any]:
    state = _load_plan_state(root)
    plan = _get_plan(state, plan_id)
    if plan.get('status') != 'confirmed':
        raise DataRecordError(f'generation plan is not confirmed: {plan_id}')
    return plan


def _require_confirmed_generation_goal(root: str, goal_id: str) -> dict[str, Any]:
    state = _load_goal_state(root)
    goal = _get_goal(state, goal_id)
    if goal.get('status') != 'confirmed':
        raise DataRecordError(f'generation goal is not confirmed: {goal_id}')
    return goal


def _case_to_record(
    case: dict[str, Any],
    *,
    global_dataset_label: str,
    index: int,
    batch_id: str,
    source_type: str,
    base_timestamp: int,
    timestamp_step_ms: int,
    default_request_id: str,
) -> tuple[dict[str, Any], list[str]]:
    warnings: list[str] = []
    turns = case.get('turns')
    if not isinstance(turns, list):
        raise DataRecordError(f'case {index} has invalid turns')

    user_indexes = [turn_index for turn_index, turn in enumerate(turns) if turn.get('role') == 'user']
    if not user_indexes:
        raise DataRecordError(f'case {index} must contain at least one 用户 line')
    final_user_index = user_indexes[-1]
    current_query = str(turns[final_user_index].get('text') or '').strip()
    if not current_query:
        raise DataRecordError(f'case {index} current 用户 line must be non-empty')

    raw_target, prefixed_complex = _split_complex_prefixed_target(str(case.get('target') or '').strip())
    target = _canonical_target(raw_target)
    if not target:
        raise DataRecordError(f'case {index} target is required')
    complex_value = _parse_complex(
        case.get('complex'),
        default=False if prefixed_complex is None else prefixed_complex,
    )

    prev_session = _build_prev_session(turns[:final_user_index], case_index=index)
    if len(prev_session) > 10:
        warnings.append(f'case {index} prev_session has more than 10 turns; keeping the latest 10')
        prev_session = prev_session[-10:]

    request_id = str(case.get('request_id') or default_request_id).strip()
    case_timestamp = _optional_int(case.get('timestamp'))
    turn_timestamp = case_timestamp if case_timestamp is not None else base_timestamp + (index - 1) * timestamp_step_ms * 20
    first_prev_timestamp = turn_timestamp - len(prev_session) * timestamp_step_ms
    for prev_index, item in enumerate(prev_session):
        item['timestamp'] = first_prev_timestamp + prev_index * timestamp_step_ms

    dataset_label = str(case.get('dataset_label') or global_dataset_label or '').strip()
    record = {
        'record_id': _record_id(source_type=source_type, batch_id=batch_id, request_id=request_id, index=index),
        'source': {
            'type': source_type,
            'request_id': request_id,
            'timestamp': turn_timestamp,
        },
        'turn': {
            'query': current_query,
            'timestamp': turn_timestamp,
        },
        'prev_session': prev_session,
        'context': {},
        'label': {
            'dataset_label': dataset_label,
            'target': target,
            'target_type': _target_type(target),
        },
        'dimensions': {
            'complex': complex_value,
        },
        'meta': {
            'case_name': str(case.get('case_name') or '').strip(),
            'notes': str(case.get('notes') or '').strip(),
        },
    }
    return record, warnings


def _build_prev_session(turns: list[dict[str, Any]], *, case_index: int) -> list[dict[str, Any]]:
    prev_session: list[dict[str, Any]] = []
    index = 0
    while index < len(turns):
        turn = turns[index]
        if turn.get('role') != 'user':
            raise DataRecordError(f'case {case_index} prev_session must start with 用户 before 小爱')
        if index + 1 >= len(turns) or turns[index + 1].get('role') != 'assistant':
            raise DataRecordError(f'case {case_index} each previous 用户 line must be followed by 小爱')
        query = str(turn.get('text') or '').strip()
        tts = str(turns[index + 1].get('text') or '').strip()
        if not query or not tts:
            raise DataRecordError(f'case {case_index} previous 用户/小爱 lines must be non-empty')
        prev_session.append({'query': query, 'tts': tts})
        index += 2
    return prev_session


def _normalize_target_expression(target: str) -> str:
    """把可能带 complex 前缀的标签表达式收敛为纯 target。"""

    return _canonical_target(_strip_complex_prefix(target))


def _strip_complex_prefix(target: str) -> str:
    stripped_target, _complex_value = _split_complex_prefixed_target(target)
    return stripped_target


def _split_complex_prefixed_target(target: str) -> tuple[str, bool | None]:
    lines = target.strip().splitlines()
    if not lines:
        return '', None
    first_line = lines[0].strip()
    match = re.fullmatch(r'complex\s*=\s*(.+)', first_line, flags=re.IGNORECASE)
    if not match:
        return target.strip(), None
    complex_value = _parse_complex(match.group(1), default=False)
    return '\n'.join(lines[1:]).strip(), complex_value


def _parse_complex(value: Any, *, default: bool) -> bool:
    if value is None or value == '':
        return default
    if isinstance(value, bool):
        return value
    text = str(value).strip().lower()
    if text in {'true', '1', 'yes', 'y', '是', '复杂', 'complex'}:
        return True
    if text in {'false', '0', 'no', 'n', '否', '不复杂', '简单', 'simple'}:
        return False
    raise DataRecordError('complex must be a boolean value such as true/false')


def _complex_literal(value: bool) -> str:
    return 'true' if value else 'false'


def _eval_complex_literal(value: bool) -> str:
    return 'TRUE' if value else 'FALSE'


def _record_target(record: dict[str, Any]) -> str:
    label = record.get('label') if isinstance(record.get('label'), dict) else {}
    return _normalize_target_expression(str(label.get('target') or ''))


def _record_complex(record: dict[str, Any], *, default: bool = False) -> bool:
    dimensions = record.get('dimensions')
    if isinstance(dimensions, dict) and 'complex' in dimensions:
        return _parse_complex(dimensions.get('complex'), default=default)
    label = record.get('label') if isinstance(record.get('label'), dict) else {}
    if 'complex' in label:
        return _parse_complex(label.get('complex'), default=default)
    _target, prefixed_complex = _split_complex_prefixed_target(str(label.get('target') or ''))
    if prefixed_complex is not None:
        return prefixed_complex
    return default


def _combined_function_label(record: dict[str, Any], *, default_complex: bool = False) -> str:
    target = _record_target(record)
    return f'complex={_complex_literal(_record_complex(record, default=default_complex))}\n{target}'


def _target_type(target: str) -> str:
    target = _strip_complex_prefix(target)
    if re.match(r'^Agent\s*\(\s*tag\s*=', target):
        return 'agent'
    if target:
        return 'function'
    return 'unknown'


def _canonical_target(target: str) -> str:
    """规范常见 Agent 标签写法，允许草稿里用单引号降低 JSON 转义风险。"""

    target = _strip_complex_prefix(target)
    match = re.fullmatch(r'Agent\s*\(\s*tag\s*=\s*([\'"])(.+?)\1\s*\)', target)
    if not match:
        return target
    tag = match.group(2).strip()
    escaped = tag.replace('\\', '\\\\').replace('"', '\\"')
    return f'Agent(tag="{escaped}")'


def _record_id(*, source_type: str, batch_id: str, request_id: str, index: int) -> str:
    if source_type == 'generated':
        return f'gen_{_safe_token(batch_id)}_{index:06d}'
    if source_type == 'online':
        return f'online_{_safe_token(request_id)}_{index:06d}'
    return f'{_safe_token(source_type)}_{_safe_token(batch_id)}_{index:06d}'


def _safe_token(value: str) -> str:
    token = re.sub(r'[^A-Za-z0-9_.-]+', '_', value.strip())
    return token.strip('_') or 'unknown'


def _resolve_output_path(root: str, output_path: str) -> Path:
    root_path = Path(root).resolve()
    candidate = Path(output_path).expanduser()
    path = candidate if candidate.is_absolute() else root_path / candidate
    resolved = path.resolve(strict=False)
    try:
        resolved.relative_to(root_path)
    except ValueError as exc:
        raise DataRecordError(f'output_path escapes workspace root: {output_path}') from exc
    return resolved


def _optional_int(value: Any) -> int | None:
    if value is None or value == '':
        return None
    if isinstance(value, bool):
        raise DataRecordError('timestamp must be an integer')
    try:
        return int(value)
    except (TypeError, ValueError) as exc:
        raise DataRecordError('timestamp must be an integer') from exc


def _non_empty_str(value: Any) -> bool:
    return isinstance(value, str) and bool(value.strip())


def _int_like(value: Any) -> bool:
    return isinstance(value, int) and not isinstance(value, bool)


def _issue(path: str, message: str) -> dict[str, str]:
    return {'path': path, 'message': message}


def records_from_tool_argument(value: Any) -> list[dict[str, Any]]:
    if isinstance(value, str):
        decoded = json.loads(value)
    else:
        decoded = value
    if not isinstance(decoded, list):
        raise DataRecordError('records must be a JSON array')
    return decoded