YG_FT/backend/tests/test_data_process_generation.py

from __future__ import annotations

import json

import httpx
import pytest

from app.modules.data_process.generation import (
    ModelGenerationError,
    chat_completions_url,
    generate_model_records,
)


def test_chat_completions_url_accepts_host_base_and_complete_url() -> None:
    assert chat_completions_url("www.caoxiaozhu.com") == (
        "https://www.caoxiaozhu.com/v1/chat/completions"
    )
    assert chat_completions_url("https://model.example/v1") == (
        "https://model.example/v1/chat/completions"
    )
    complete = "https://model.example/openai/v1/chat/completions"
    assert chat_completions_url(complete) == complete


def test_generate_model_records_uses_prompt_auth_and_stable_split() -> None:
    requests: list[httpx.Request] = []
    progress_updates: list[tuple[int, int]] = []

    def handler(request: httpx.Request) -> httpx.Response:
        requests.append(request)
        payload = json.loads(request.content)
        assert payload["model"] == "qwen-plus"
        assert payload["response_format"] == {"type": "json_object"}
        assert "客户反馈页面加载慢" in payload["messages"][1]["content"]
        assert "你正在生成标准监督微调问答数据" in payload["messages"][0]["content"]
        assert "禁止输出分析、推理过程" in payload["messages"][0]["content"]
        assert "真实用户自然提出的问题" in payload["messages"][0]["content"]
        assert "模板化开头" in payload["messages"][0]["content"]
        return httpx.Response(
            200,
            json={
                "choices": [
                    {
                        "message": {
                            "content": json.dumps(
                                {
                                    "items": [
                                        {
                                            "instruction": "请生成简洁客服回复",
                                            "input": "客户反馈页面加载慢",
                                            "output": "已收到反馈，我们正在排查。",
                                        }
                                    ]
                                },
                                ensure_ascii=False,
                            )
                        }
                    }
                ]
            },
        )

    client = httpx.Client(transport=httpx.MockTransport(handler))
    records = generate_model_records(
        [{"id": "preview-1", "edited_content": "客户反馈页面加载慢"}],
        model={
            "name": "Qwen",
            "online_model_name": "qwen-plus",
            "api_url": "model.example",
            "api_key": "test-secret",
        },
        config={
            "generation_prompt": "请处理：{{ content }}",
            "json_mode": True,
            "temperature": 0.2,
            "max_tokens": 512,
        },
        task_id="task-1",
        split={"train": 100, "validation": 0, "test": 0},
        qa_pairs_per_item=1,
        client=client,
        on_progress=lambda processed, total: progress_updates.append((processed, total)),
    )

    assert len(records) == 1
    assert records[0]["status"] == "valid"
    assert records[0]["split"] == "train"
    assert requests[0].headers["Authorization"] == "Bearer test-secret"
    assert progress_updates == [(1, 1)]


def test_generate_model_records_falls_back_to_rich_default_prompt() -> None:
    captured: dict[str, str] = {}

    def handler(request: httpx.Request) -> httpx.Response:
        payload = json.loads(request.content)
        captured["system"] = payload["messages"][0]["content"]
        captured["user"] = payload["messages"][1]["content"]
        return httpx.Response(
            200,
            json={
                "choices": [
                    {
                        "message": {
                            "content": json.dumps(
                                {
                                    "items": [
                                        {
                                            "instruction": "平台如何控制不同角色的菜单可见性？",
                                            "input": "",
                                            "output": "按角色分配权限。",
                                        }
                                    ]
                                },
                                ensure_ascii=False,
                            )
                        }
                    }
                ]
            },
        )

    client = httpx.Client(transport=httpx.MockTransport(handler))
    records = generate_model_records(
        [{"id": "preview-1", "edited_content": "平台按角色分配菜单权限。"}],
        model={
            "name": "Qwen",
            "online_model_name": "qwen-plus",
            "api_url": "model.example",
        },
        config={},
        task_id="task-fallback",
        split={"train": 100, "validation": 0, "test": 0},
        qa_pairs_per_item=1,
        client=client,
    )

    assert len(records) == 1
    assert records[0]["status"] == "valid"
    assert "数据生成专家" in captured["system"]
    assert "真实用户自然提出的问题" in captured["system"]
    assert "平台按角色分配菜单权限。" in captured["user"]


def test_generate_model_records_builds_native_dpo_pair() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        payload = json.loads(request.content)
        system_prompt = payload["messages"][0]["content"]
        assert '"chosen"' in system_prompt
        assert '"rejected"' in system_prompt
        assert "直接偏好优化" in system_prompt
        return httpx.Response(
            200,
            json={
                "choices": [{
                    "message": {
                        "content": json.dumps({
                            "items": [{
                                "instruction": "系统如何处理扫描 PDF？",
                                "input": "",
                                "chosen": "仅在没有文本层时调用 OCR，并保留页码。",
                                "rejected": "所有 PDF 都重复执行 OCR。",
                            }],
                        }, ensure_ascii=False),
                    },
                }],
            },
        )

    records = generate_model_records(
        [{"id": "preview-dpo", "edited_content": "扫描 PDF 缺少文本层时执行 OCR。"}],
        model={"name": "model", "api_url": "https://model.example/v1"},
        config={"output_type": "dpo", "generation_retries": 0},
        task_id="task-dpo",
        split={"train": 100, "validation": 0, "test": 0},
        qa_pairs_per_item=1,
        client=httpx.Client(transport=httpx.MockTransport(handler)),
    )

    assert records[0]["status"] == "valid"
    assert records[0]["chosen"] == "仅在没有文本层时调用 OCR,并保留页码。"
    assert records[0]["rejected"] == "所有 PDF 都重复执行 OCR。"
    assert records[0]["output"] == records[0]["chosen"]


def test_generate_model_records_rejects_equal_dpo_pair() -> None:
    def handler(_request: httpx.Request) -> httpx.Response:
        return httpx.Response(
            200,
            json={"choices": [{"message": {"content": json.dumps({
                "items": [{
                    "instruction": "问题",
                    "chosen": "相同回答",
                    "rejected": "相同回答",
                }],
            }, ensure_ascii=False)}}]},
        )

    records = generate_model_records(
        [{"id": "preview-dpo-invalid", "edited_content": "来源"}],
        model={"name": "model", "api_url": "https://model.example/v1"},
        config={"output_type": "dpo", "generation_retries": 0},
        task_id="task-dpo-invalid",
        split={"train": 100, "validation": 0, "test": 0},
        qa_pairs_per_item=1,
        client=httpx.Client(transport=httpx.MockTransport(handler)),
    )

    assert records[0]["status"] == "invalid"
    assert "chosen equals rejected" in records[0]["error"]


def test_minimax_m3_uses_split_reasoning_and_completion_token_budget() -> None:
    requests: list[dict[str, object]] = []

    def handler(request: httpx.Request) -> httpx.Response:
        payload = json.loads(request.content)
        requests.append(payload)
        return httpx.Response(
            200,
            json={
                "choices": [{
                    "finish_reason": "stop",
                    "message": {
                        "reasoning_content": "模型内部思考不应混入业务 JSON",
                        "content": json.dumps({
                            "items": [{
                                "instruction": "申请编号有什么作用？",
                                "reasoning": "来源说明它用于标识报销申请。",
                                "answer": "它用于唯一标识一笔报销申请。",
                            }],
                        }, ensure_ascii=False),
                    },
                }],
                "output_sensitive": False,
                "base_resp": {"status_code": 0, "status_msg": ""},
            },
        )

    records = generate_model_records(
        [{"id": "preview-minimax", "edited_content": "申请编号用于标识报销申请。"}],
        model={
            "name": "MiniMax",
            "online_model_name": "MiniMax-M3",
            "api_url": "https://api.minimaxi.com/v1",
        },
        config={
            "output_type": "reasoning",
            "json_mode": True,
            "max_tokens": 1024,
            "generation_retries": 0,
        },
        task_id="task-minimax",
        split={"train": 100, "validation": 0, "test": 0},
        qa_pairs_per_item=1,
        client=httpx.Client(transport=httpx.MockTransport(handler)),
    )

    assert records[0]["status"] == "valid"
    assert len(requests) == 1
    assert requests[0]["reasoning_split"] is True
    assert requests[0]["max_completion_tokens"] >= 4096
    assert "max_tokens" not in requests[0]
    assert "response_format" not in requests[0]


def test_minimax_m3_keeps_larger_configured_completion_budget() -> None:
    requests: list[dict[str, object]] = []

    def handler(request: httpx.Request) -> httpx.Response:
        requests.append(json.loads(request.content))
        return httpx.Response(
            200,
            json={
                "choices": [{
                    "message": {
                        "content": json.dumps({
                            "items": [{
                                "instruction": "问题",
                                "output": "这是满足测试要求的完整答案。",
                            }],
                        }, ensure_ascii=False),
                    },
                }],
            },
        )

    generate_model_records(
        [{"id": "preview-minimax-budget", "edited_content": "来源正文"}],
        model={
            "online_model_name": "MiniMax-M3",
            "api_url": "https://api.minimax.io/v1",
        },
        config={"max_tokens": 8192, "generation_retries": 0},
        task_id="task-minimax-budget",
        split={"train": 100, "validation": 0, "test": 0},
        qa_pairs_per_item=1,
        client=httpx.Client(transport=httpx.MockTransport(handler)),
    )

    assert requests[0]["max_completion_tokens"] == 8192


def test_minimax_m3_name_on_custom_proxy_keeps_generic_openai_parameters() -> None:
    requests: list[dict[str, object]] = []

    def handler(request: httpx.Request) -> httpx.Response:
        requests.append(json.loads(request.content))
        return httpx.Response(
            200,
            json={
                "choices": [{
                    "message": {
                        "content": json.dumps({
                            "items": [{
                                "instruction": "问题",
                                "output": "这是代理服务返回的完整答案。",
                            }],
                        }, ensure_ascii=False),
                    },
                }],
            },
        )

    generate_model_records(
        [{"id": "preview-minimax-proxy", "edited_content": "来源正文"}],
        model={
            "online_model_name": "MiniMax-M3",
            "api_url": "https://model-proxy.example/v1",
        },
        config={"max_tokens": 1024, "json_mode": True, "generation_retries": 0},
        task_id="task-minimax-proxy",
        split={"train": 100, "validation": 0, "test": 0},
        qa_pairs_per_item=1,
        client=httpx.Client(transport=httpx.MockTransport(handler)),
    )

    assert requests[0]["max_tokens"] == 1024
    assert requests[0]["response_format"] == {"type": "json_object"}
    assert "reasoning_split" not in requests[0]
    assert "max_completion_tokens" not in requests[0]


def test_generate_model_records_extracts_json_surrounded_by_model_explanation() -> None:
    content = "模型结果如下：\n```json\n" + json.dumps(
        {
            "items": [{
                "instruction": "字段有什么作用？",
                "output": "该字段用于唯一标识记录。",
            }],
        },
        ensure_ascii=False,
    ) + "\n```\n生成完毕。"
    client = httpx.Client(
        transport=httpx.MockTransport(
            lambda _: httpx.Response(
                200,
                json={"choices": [{"message": {"content": content}}]},
            )
        )
    )

    records = generate_model_records(
        [{"id": "preview-explanation", "edited_content": "字段用于唯一标识记录。"}],
        model={"name": "model", "api_url": "https://model.example/v1"},
        config={"generation_retries": 0},
        task_id="task-explanation",
        split={"train": 100, "validation": 0, "test": 0},
        qa_pairs_per_item=1,
        client=client,
    )

    assert records[0]["status"] == "valid"
    assert records[0]["output"] == "该字段用于唯一标识记录。"


def test_generate_model_records_reports_token_truncation_instead_of_json_error() -> None:
    client = httpx.Client(
        transport=httpx.MockTransport(
            lambda _: httpx.Response(
                200,
                json={
                    "choices": [{
                        "finish_reason": "length",
                        "message": {"content": ""},
                    }],
                    "output_sensitive": False,
                },
            )
        )
    )

    records = generate_model_records(
        [{"id": "preview-truncated", "edited_content": "来源正文"}],
        model={"name": "model", "api_url": "https://model.example/v1"},
        config={"generation_retries": 0},
        task_id="task-truncated",
        split={"train": 100, "validation": 0, "test": 0},
        qa_pairs_per_item=1,
        client=client,
    )

    assert records[0]["status"] == "invalid"
    assert "Token" in records[0]["error"]
    assert "截断" in records[0]["error"]


def test_token_truncation_is_not_retried_even_when_json_looks_complete() -> None:
    request_count = 0
    content = json.dumps({
        "items": [{
            "instruction": "问题",
            "output": "表面完整但服务端已声明截断。",
        }],
    }, ensure_ascii=False)

    def handler(_: httpx.Request) -> httpx.Response:
        nonlocal request_count
        request_count += 1
        return httpx.Response(
            200,
            json={
                "choices": [{
                    "finish_reason": "length",
                    "message": {"content": content},
                }],
            },
        )

    records = generate_model_records(
        [{"id": "preview-length", "edited_content": "来源正文"}],
        model={"name": "model", "api_url": "https://model.example/v1"},
        config={"generation_retries": 5},
        task_id="task-length",
        split={"train": 100, "validation": 0, "test": 0},
        qa_pairs_per_item=1,
        client=httpx.Client(transport=httpx.MockTransport(handler)),
    )

    assert request_count == 1
    assert records[0]["status"] == "invalid"
    assert "finish_reason=length" in records[0]["error"]


def test_sensitive_model_response_is_not_retried_or_saved() -> None:
    request_count = 0

    def handler(_: httpx.Request) -> httpx.Response:
        nonlocal request_count
        request_count += 1
        return httpx.Response(
            200,
            json={
                "choices": [{
                    "finish_reason": "stop",
                    "message": {"content": "{}"},
                }],
                "output_sensitive": True,
                "base_resp": {"status_code": 1027, "status_msg": "output sensitive"},
            },
        )

    records = generate_model_records(
        [{"id": "preview-sensitive", "edited_content": "来源正文"}],
        model={"name": "model", "api_url": "https://model.example/v1"},
        config={"generation_retries": 5},
        task_id="task-sensitive",
        split={"train": 100, "validation": 0, "test": 0},
        qa_pairs_per_item=1,
        client=httpx.Client(transport=httpx.MockTransport(handler)),
    )

    assert request_count == 1
    assert records[0]["status"] == "invalid"
    assert "安全拦截" in records[0]["error"]
    assert "1027" in records[0]["error"]


def test_empty_model_content_can_retry_then_succeed() -> None:
    request_count = 0

    def handler(_: httpx.Request) -> httpx.Response:
        nonlocal request_count
        request_count += 1
        if request_count == 1:
            return httpx.Response(
                200,
                json={"choices": [{"finish_reason": "stop", "message": {"content": ""}}]},
            )
        return httpx.Response(
            200,
            json={
                "choices": [{
                    "finish_reason": "stop",
                    "message": {
                        "content": json.dumps({
                            "items": [{
                                "instruction": "问题",
                                "output": "第二次请求返回了完整答案。",
                            }],
                        }, ensure_ascii=False),
                    },
                }],
            },
        )

    records = generate_model_records(
        [{"id": "preview-empty-retry", "edited_content": "来源正文"}],
        model={"name": "model", "api_url": "https://model.example/v1"},
        config={"generation_retries": 1},
        task_id="task-empty-retry",
        split={"train": 100, "validation": 0, "test": 0},
        qa_pairs_per_item=1,
        client=httpx.Client(transport=httpx.MockTransport(handler)),
    )

    assert request_count == 2
    assert records[0]["status"] == "valid"


def test_multiple_top_level_json_documents_are_rejected_as_ambiguous() -> None:
    first = json.dumps({
        "items": [{"instruction": "问题一", "output": "答案一"}],
    }, ensure_ascii=False)
    second = json.dumps({
        "items": [{"instruction": "问题二", "output": "答案二"}],
    }, ensure_ascii=False)
    client = httpx.Client(
        transport=httpx.MockTransport(
            lambda _: httpx.Response(
                200,
                json={"choices": [{"message": {"content": f"{first}\n{second}"}}]},
            )
        )
    )

    records = generate_model_records(
        [{"id": "preview-ambiguous", "edited_content": "来源正文"}],
        model={"name": "model", "api_url": "https://model.example/v1"},
        config={"generation_retries": 0},
        task_id="task-ambiguous",
        split={"train": 100, "validation": 0, "test": 0},
        qa_pairs_per_item=1,
        client=client,
    )

    assert records[0]["status"] == "invalid"
    assert "多个 JSON" in records[0]["error"]


def test_generate_model_records_builds_reasoning_output_with_think_tags() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        payload = json.loads(request.content)
        system_prompt = payload["messages"][0]["content"]
        assert '"reasoning":"...","answer":"..."' in system_prompt
        assert "你正在生成用于训练推理模型的思维链数据" in system_prompt
        assert "推理详细程度为“普通”" in system_prompt
        assert "系统会在保存时统一组装" in system_prompt
        content = "<think>模型接口自己的分析</think>" + json.dumps(
            {
                "items": [
                    {
                        "instruction": "计算两项费用合计",
                        "input": "交通费 30 元，餐费 20 元",
                        "reasoning": "先识别两项费用，再计算 30 + 20。",
                        "answer": "合计 50 元。",
                    }
                ]
            },
            ensure_ascii=False,
        )
        return httpx.Response(
            200,
            json={"choices": [{"message": {"content": content}}]},
        )

    records = generate_model_records(
        [{"id": "preview-reasoning", "edited_content": "交通费 30 元，餐费 20 元"}],
        model={"name": "model", "api_url": "https://model.example/v1"},
        config={"output_type": "reasoning"},
        task_id="task-reasoning",
        split={"train": 100, "validation": 0, "test": 0},
        qa_pairs_per_item=1,
        client=httpx.Client(transport=httpx.MockTransport(handler)),
    )

    assert records[0]["status"] == "valid"
    assert records[0]["output"] == (
        "<think>\n先识别两项费用,再计算 30 + 20。\n</think>\n合计 50 元。"
    )


def test_generate_model_records_uses_detailed_reasoning_instruction() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        system_prompt = json.loads(request.content)["messages"][0]["content"]
        assert "推理详细程度为“详细”" in system_prompt
        assert "完整展开问题条件、来源依据、中间计算或推导" in system_prompt
        return httpx.Response(
            200,
            json={
                "choices": [
                    {
                        "message": {
                            "content": json.dumps(
                                {
                                    "items": [
                                        {
                                            "instruction": "计算报销总额",
                                            "reasoning": "条件为交通费 30 元和餐费 20 元。分别核对后相加，30 + 20 = 50。",
                                            "answer": "报销总额为 50 元。",
                                        }
                                    ]
                                },
                                ensure_ascii=False,
                            )
                        }
                    }
                ]
            },
        )

    records = generate_model_records(
        [{"id": "preview-detailed", "edited_content": "交通费 30 元，餐费 20 元"}],
        model={"name": "model", "api_url": "https://model.example/v1"},
        config={"output_type": "reasoning", "reasoning_detail": "detailed"},
        task_id="task-detailed",
        split={"train": 100, "validation": 0, "test": 0},
        qa_pairs_per_item=1,
        client=httpx.Client(transport=httpx.MockTransport(handler)),
    )

    assert records[0]["status"] == "valid"
    assert "分别核对后相加" in records[0]["output"]


def test_generate_model_records_marks_reasoning_without_reasoning_field_invalid() -> None:
    response = {
        "choices": [
            {
                "message": {
                    "content": json.dumps(
                        {
                            "items": [
                                {
                                    "instruction": "问题",
                                    "answer": "只有最终答案",
                                }
                            ]
                        },
                        ensure_ascii=False,
                    )
                }
            }
        ]
    }
    client = httpx.Client(
        transport=httpx.MockTransport(lambda _: httpx.Response(200, json=response))
    )

    records = generate_model_records(
        [{"id": "preview-missing-reasoning", "edited_content": "来源正文"}],
        model={"name": "model", "api_url": "https://model.example/v1"},
        config={"output_type": "reasoning"},
        task_id="task-missing-reasoning",
        split={"train": 100, "validation": 0, "test": 0},
        qa_pairs_per_item=1,
        client=client,
    )

    assert records[0]["status"] == "invalid"
    assert records[0]["output"] == "只有最终答案"
    assert "reasoning" in records[0]["error"]


def test_standard_output_removes_model_think_block() -> None:
    content = json.dumps(
        {
            "items": [
                {
                    "instruction": "问题",
                    "output": "<think>不应保存的分析</think>最终答案",
                }
            ]
        },
        ensure_ascii=False,
    )
    client = httpx.Client(
        transport=httpx.MockTransport(
            lambda _: httpx.Response(
                200,
                json={"choices": [{"message": {"content": content}}]},
            )
        )
    )

    records = generate_model_records(
        [{"id": "preview-standard", "edited_content": "来源正文"}],
        model={"name": "model", "api_url": "https://model.example/v1"},
        config={"output_type": "standard"},
        task_id="task-standard",
        split={"train": 100, "validation": 0, "test": 0},
        qa_pairs_per_item=1,
        client=client,
    )

    assert records[0]["output"] == "最终答案"


def test_generate_model_records_keeps_partial_failure_for_manual_repair() -> None:
    client = httpx.Client(
        transport=httpx.MockTransport(
            lambda _: httpx.Response(200, json={"choices": [{"message": {"content": "not-json"}}]})
        )
    )
    records = generate_model_records(
        [{"id": "preview-1", "edited_content": "来源正文"}],
        model={"name": "model", "api_url": "https://model.example/v1"},
        config={"generation_retries": 1},
        task_id="task-1",
        split={"train": 80, "validation": 10, "test": 10},
        qa_pairs_per_item=1,
        client=client,
    )

    assert len(records) == 1
    assert records[0]["status"] == "invalid"
    assert records[0]["error"]


def test_generate_model_records_batches_fifty_results_with_unique_ids() -> None:
    requests: list[httpx.Request] = []

    def handler(request: httpx.Request) -> httpx.Response:
        requests.append(request)
        batch_start = (len(requests) - 1) * 10 + 1
        batch_end = batch_start + 9
        payload = json.loads(request.content)
        system_prompt = payload["messages"][0]["content"]
        assert "items 必须包含 10 条" in system_prompt
        assert f"第 {batch_start}-{batch_end} 条" in system_prompt
        return httpx.Response(
            200,
            json={
                "choices": [
                    {
                        "message": {
                            "content": json.dumps(
                                {
                                    "items": [
                                        {
                                            "instruction": "同一问题",
                                            "input": "来源正文",
                                            "output": "同一答案",
                                        }
                                        for _ in range(batch_start, batch_end + 1)
                                    ]
                                },
                                ensure_ascii=False,
                            )
                        }
                    }
                ]
            },
        )

    records = generate_model_records(
        [{"id": "preview-50", "edited_content": "来源正文"}],
        model={"name": "model", "api_url": "https://model.example/v1"},
        config={},
        task_id="task-50",
        split={"train": 100, "validation": 0, "test": 0},
        qa_pairs_per_item=50,
        client=httpx.Client(transport=httpx.MockTransport(handler)),
    )

    assert len(requests) == 5
    assert len(records) == 50
    assert len({record["id"] for record in records}) == 50
    assert {record["instruction"] for record in records} == {"同一问题"}
    assert all(record["status"] == "valid" for record in records)


def test_generate_model_records_preserves_successful_batches_when_one_fails() -> None:
    request_count = 0

    def handler(_: httpx.Request) -> httpx.Response:
        nonlocal request_count
        request_count += 1
        if request_count == 2:
            return httpx.Response(500)
        return httpx.Response(
            200,
            json={
                "choices": [
                    {
                        "message": {
                            "content": json.dumps(
                                {
                                    "items": [
                                        {
                                            "instruction": f"问题 {index}",
                                            "output": f"答案 {index}",
                                        }
                                        for index in range(1, 11)
                                    ]
                                },
                                ensure_ascii=False,
                            )
                        }
                    }
                ]
            },
        )

    records = generate_model_records(
        [{"id": "preview-partial", "edited_content": "来源正文"}],
        model={"name": "model", "api_url": "https://model.example/v1"},
        config={"generation_retries": 0},
        task_id="task-partial",
        split={"train": 100, "validation": 0, "test": 0},
        qa_pairs_per_item=20,
        client=httpx.Client(transport=httpx.MockTransport(handler)),
    )

    assert len(records) == 11
    assert sum(record["status"] == "valid" for record in records) == 10
    failed = next(record for record in records if record["status"] == "invalid")
    assert "第 11-20 条" in failed["instruction"]
    assert len({record["id"] for record in records}) == len(records)


def test_generate_model_records_retries_short_batch_then_marks_it_invalid() -> None:
    request_count = 0

    def handler(_: httpx.Request) -> httpx.Response:
        nonlocal request_count
        request_count += 1
        return httpx.Response(
            200,
            json={
                "choices": [
                    {
                        "message": {
                            "content": json.dumps(
                                {
                                    "items": [
                                        {
                                            "instruction": "只有一条",
                                            "output": "不足本批要求数量",
                                        }
                                    ]
                                },
                                ensure_ascii=False,
                            )
                        }
                    }
                ]
            },
        )

    records = generate_model_records(
        [{"id": "preview-short", "edited_content": "来源正文"}],
        model={"name": "model", "api_url": "https://model.example/v1"},
        config={"generation_retries": 1},
        task_id="task-short",
        split={"train": 100, "validation": 0, "test": 0},
        qa_pairs_per_item=10,
        client=httpx.Client(transport=httpx.MockTransport(handler)),
    )

    assert request_count == 2
    assert len(records) == 1
    assert records[0]["status"] == "invalid"
    assert "expected 10, got 1" in records[0]["error"]


def test_generate_model_records_does_not_retry_non_retryable_http_errors() -> None:
    request_count = 0

    def handler(_: httpx.Request) -> httpx.Response:
        nonlocal request_count
        request_count += 1
        return httpx.Response(401, json={"error": {"message": "unauthorized"}})

    records = generate_model_records(
        [{"id": "preview-auth", "edited_content": "来源内容"}],
        model={
            "api_url": "https://model.example/v1",
            "online_model_name": "test-model",
            "api_key": "invalid",
        },
        config={"generation_retries": 5},
        task_id="task-auth",
        split={"train": 100, "validation": 0, "test": 0},
        qa_pairs_per_item=1,
        client=httpx.Client(transport=httpx.MockTransport(handler)),
    )

    assert request_count == 1
    assert records[0]["status"] == "invalid"
    assert "401" in records[0]["error"]


@pytest.mark.parametrize("status_code", [408, 425, 429, 500])
def test_generate_model_records_retries_retryable_http_statuses(
    status_code: int,
) -> None:
    request_count = 0

    def handler(_: httpx.Request) -> httpx.Response:
        nonlocal request_count
        request_count += 1
        if request_count == 1:
            return httpx.Response(status_code)
        return httpx.Response(
            200,
            json={
                "choices": [{
                    "message": {
                        "content": json.dumps({
                            "items": [{
                                "instruction": "来源内容是什么？",
                                "output": "这是用于验证可重试错误的来源内容。",
                            }],
                        }, ensure_ascii=False),
                    },
                }],
            },
        )

    records = generate_model_records(
        [{"id": "preview-retryable", "edited_content": "来源内容"}],
        model={"name": "model", "api_url": "https://model.example/v1"},
        config={"generation_retries": 1},
        task_id="task-retryable",
        split={"train": 100, "validation": 0, "test": 0},
        qa_pairs_per_item=1,
        client=httpx.Client(transport=httpx.MockTransport(handler)),
    )

    assert request_count == 2
    assert records[0]["status"] == "valid"


def test_generate_model_records_retries_transient_network_errors() -> None:
    request_count = 0

    def handler(request: httpx.Request) -> httpx.Response:
        nonlocal request_count
        request_count += 1
        if request_count == 1:
            raise httpx.ConnectError("temporary connection failure", request=request)
        return httpx.Response(
            200,
            json={
                "choices": [{
                    "message": {
                        "content": json.dumps({
                            "items": [{
                                "instruction": "网络恢复了吗？",
                                "output": "临时连接错误后，第二次模型请求已经成功。",
                            }],
                        }, ensure_ascii=False),
                    },
                }],
            },
        )

    records = generate_model_records(
        [{"id": "preview-network", "edited_content": "网络重试来源"}],
        model={"name": "model", "api_url": "https://model.example/v1"},
        config={"generation_retries": 1},
        task_id="task-network",
        split={"train": 100, "validation": 0, "test": 0},
        qa_pairs_per_item=1,
        client=httpx.Client(transport=httpx.MockTransport(handler)),
    )

    assert request_count == 2
    assert records[0]["status"] == "valid"


@pytest.mark.parametrize("qa_pairs_per_item", [0, 51])
def test_generate_model_records_rejects_out_of_range_count(
    qa_pairs_per_item: int,
) -> None:
    with pytest.raises(ModelGenerationError, match=r"\[1, 50\]"):
        generate_model_records(
            [],
            model={"name": "model", "api_url": "https://model.example/v1"},
            config={},
            task_id="task-invalid",
            split={"train": 100, "validation": 0, "test": 0},
            qa_pairs_per_item=qa_pairs_per_item,
        )


def test_generate_model_records_rejects_unknown_output_type() -> None:
    with pytest.raises(ModelGenerationError, match="output_type"):
        generate_model_records(
            [],
            model={"name": "model", "api_url": "https://model.example/v1"},
            config={"output_type": "unknown"},
            task_id="task-invalid-output",
            split={"train": 100, "validation": 0, "test": 0},
            qa_pairs_per_item=1,
        )


def test_generate_model_records_rejects_unknown_reasoning_detail() -> None:
    with pytest.raises(ModelGenerationError, match="reasoning_detail"):
        generate_model_records(
            [],
            model={"name": "model", "api_url": "https://model.example/v1"},
            config={"output_type": "reasoning", "reasoning_detail": "verbose"},
            task_id="task-invalid-reasoning-detail",
            split={"train": 100, "validation": 0, "test": 0},
            qa_pairs_per_item=1,
        )