From 6072ee7d0bccd5366cd94ed5a2d7efd66e7ff7b6 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 7 Jul 2026 10:12:15 -0400 Subject: [PATCH] feat(evolution): evolve.py validation + helpers (#9) --- core/evolution/evolve.py | 724 ++++++++++++++++++++++++++++++++++++++ tests/unit/test_evolve.py | 659 ++++++++++++++++++++++++++++++++++ 2 files changed, 1383 insertions(+) create mode 100644 core/evolution/evolve.py create mode 100644 tests/unit/test_evolve.py diff --git a/core/evolution/evolve.py b/core/evolution/evolve.py new file mode 100644 index 0000000..09ce49e --- /dev/null +++ b/core/evolution/evolve.py @@ -0,0 +1,724 @@ +"""进化引擎辅助函数与验证逻辑。 + +验证（validate_skill / validate_system / validate_tool）、受保护区构建、 +编辑预算退火、rank-and-clip 裁剪、格式化工具等纯/准纯函数。 +Task 8 将在此基础上添加进化入口（evolve_skill / evolve_system / evolve_tool）。 + +不依赖 app/ 或 adapters/（LLMProvider 通过 core.protocols 注入）。 +""" + +from __future__ import annotations + +import json +import re +from dataclasses import asdict, dataclass, field +from typing import TYPE_CHECKING, Any + +from loguru import logger + +from core.evolution.patch import ( + APPENDIX_END, + APPENDIX_START, + momentum_region_bounds, +) + +if TYPE_CHECKING: + from core.evolution.protocols import SkillStore + from core.evolution.types import RejectedEdit + from core.protocols import LLMProvider + +# ========================================================================= +# 0. 局部类型 +# ========================================================================= + + +@dataclass +class ValidationResult: + """格式验证的结果。 + + 属性: + passed: 验证是否通过。 + errors: 失败原因列表；passed=True 时为空。 + """ + + passed: bool + errors: list[str] = field(default_factory=list) + + +# ========================================================================= +# A. 内部辅助函数 +# ========================================================================= + + +def _parse_frontmatter(text: str) -> dict[str, str] | None: + """解析 YAML frontmatter，失败返回 None。 + + 参数: + text: Markdown 文件全文。 + + 返回: + frontmatter 字典，无有效 frontmatter 时返回 None。 + """ + import yaml as _yaml + + match = re.match(r"^---\n(.*?)\n---", text, re.DOTALL) + if not match: + return None + try: + return _yaml.safe_load(match.group(1)) + except _yaml.YAMLError: + return None + + +def _strip_appendix_region(text: str) -> str: + """剥离 appendix 受保护区（含 marker），返回其余正文。 + + 宽松语义：APPENDIX_START / APPENDIX_END 任一缺失即当作「无区」原样返回，不报错。 + """ + if APPENDIX_START in text and APPENDIX_END in text: + head, rest = text.split(APPENDIX_START, 1) + _, tail = rest.split(APPENDIX_END, 1) + return head.rstrip() + tail + return text + + +def _strip_momentum_region(text: str) -> str: + """剥离 momentum 受保护区（含 marker），返回其余正文。 + + 严格语义：委托 momentum_region_bounds 做配对检测， + marker 损坏/不配对时 raise ValueError。 + + 异常: + ValueError: momentum marker 损坏/不配对。 + """ + bounds = momentum_region_bounds(text) + if bounds is None: + return text + start, end = bounds + return text[:start].rstrip() + text[end:] + + +def _strip_protected_regions(text: str) -> str: + """剥离 appendix + momentum 两个受保护区，返回正文部分。 + + 先 appendix（宽松），再 momentum（严格）——顺序有关： + appendix 宽松剥离不会误杀 momentum marker。 + + 异常: + ValueError: momentum marker 损坏/不配对。 + """ + text = _strip_appendix_region(text) + text = _strip_momentum_region(text) + return text + + +def _check_length(original: str, evolved: str) -> list[str]: + """检查改写后长度是否在 [0.3x, 2.0x] 范围内。 + + 仅比正文，剔除 appendix + momentum 两区。orig_len==0 时跳过。 + + 参数: + original: 改写前全文。 + evolved: 改写后全文。 + + 返回: + 错误消息列表（空列表表示通过）。 + """ + errors: list[str] = [] + orig_body = _strip_protected_regions(original) + evol_body = _strip_protected_regions(evolved) + orig_len = len(orig_body) + if orig_len == 0: + return errors + ratio = len(evol_body) / orig_len + evol_len = len(evol_body) + if ratio > 2.0: + errors.append( + f"长度超限: {evol_len} 字符是原文 {orig_len} 的 {ratio:.1f} 倍 (上限 2.0)" + ) + if ratio < 0.3: + errors.append( + f"长度不足: {evol_len} 字符是原文 {orig_len} 的 {ratio:.1f} 倍 (下限 0.3)" + ) + return errors + + +def _check_code_blocks(text: str) -> list[str]: + """检查代码块是否闭合。 + + 参数: + text: 待检查的文本。 + + 返回: + 错误消息列表（空列表表示通过）。 + """ + count = text.count("```") + if count % 2 != 0: + return [f"Markdown 格式错误: 代码块未闭合 (``` 出现 {count} 次)"] + return [] + + +def _extract_section(text: str, heading: str) -> str | None: + """提取 ## heading 到下一个 ## 之间的文本。 + + 参数: + text: Markdown 全文。 + heading: 二级标题名。 + + 返回: + 该 section 的完整文本（含标题行），未找到时返回 None。 + """ + pattern = rf"(## {re.escape(heading)}.*?)(?=\n## |\Z)" + match = re.search(pattern, text, re.DOTALL) + return match.group(1).strip() if match else None + + +# ========================================================================= +# B. 受保护区构建 +# ========================================================================= + + +def _appendix_span(content: str) -> str: + """返回 appendix 受保护区整段（含 marker）；不存在返回空串。 + + 参数: + content: 文本全文。 + + 返回: + appendix 区的完整文本（含 marker），或空串。 + """ + if APPENDIX_START in content and APPENDIX_END in content: + start = content.index(APPENDIX_START) + end = content.index(APPENDIX_END) + len(APPENDIX_END) + return content[start:end] + return "" + + +def _momentum_span(content: str) -> str: + """返回 momentum 受保护区整段（含 marker）；不存在返回空串。 + + 委托 momentum_region_bounds 做配对检测： + marker 损坏/不配对时由其 raise ValueError。 + + 参数: + content: 文本全文。 + + 返回: + momentum 区的完整文本（含 marker），或空串。 + + 异常: + ValueError: momentum marker 损坏/不配对。 + """ + bounds = momentum_region_bounds(content) + if bounds is None: + return "" + start, end = bounds + return content[start:end] + + +def _skill_protected_spans(text: str) -> list[str]: + """Skill 冻结块：frontmatter + appendix 区 + momentum 区（各项可选）。 + + 参数: + text: Skill 文件全文。 + + 返回: + 冻结文本块列表。 + """ + spans: list[str] = [] + match = re.match(r"^---\n.*?\n---", text, re.DOTALL) + if match: + spans.append(match.group(0)) + appendix = _appendix_span(text) + if appendix: + spans.append(appendix) + momentum = _momentum_span(text) + if momentum: + spans.append(momentum) + return spans + + +def _system_protected_spans(text: str) -> list[str]: + """System Prompt 冻结块：能力边界 / 输出格式 / 视频树结构 三段 + appendix 区。 + + 参数: + text: system.md 全文。 + + 返回: + 冻结文本块列表。 + """ + spans: list[str] = [ + section + for section in ( + _extract_section(text, name) + for name in ("能力边界", "输出格式", "视频树结构") + ) + if section + ] + appendix = _appendix_span(text) + if appendix: + spans.append(appendix) + return spans + + +def _tool_protected_spans(text: str) -> list[str]: + """Tool Prompt 冻结块：输出格式段 + appendix 区。 + + 参数: + text: Tool Prompt 全文。 + + 返回: + 冻结文本块列表。 + """ + spans: list[str] = [] + section = _extract_section(text, "输出格式") + if section: + spans.append(section) + appendix = _appendix_span(text) + if appendix: + spans.append(appendix) + return spans + + +# ========================================================================= +# C. 验证函数 +# ========================================================================= + + +def validate_skill(original: str, evolved: str) -> ValidationResult: + """校验 Skill 改写结果。 + + 检查项: frontmatter 三字段保留（name / description / task_type）、 + 长度比在 [0.3, 2.0]、代码块闭合。 + + 参数: + original: 改写前的 Skill 文件全文。 + evolved: 改写后的 Skill 文件全文。 + + 返回: + ValidationResult 实例。 + """ + errors: list[str] = [] + orig_fm = _parse_frontmatter(original) + evol_fm = _parse_frontmatter(evolved) + if orig_fm is None: + errors.append("原文缺少有效 frontmatter") + elif evol_fm is None: + errors.append("改写后缺少有效 frontmatter") + else: + for key in ("name", "description", "task_type"): + if orig_fm.get(key) != evol_fm.get(key): + errors.append( + f"frontmatter 字段 {key} 被修改: " + f"{orig_fm.get(key)!r} → {evol_fm.get(key)!r}" + ) + errors.extend(_check_length(original, evolved)) + errors.extend(_check_code_blocks(evolved)) + return ValidationResult(passed=len(errors) == 0, errors=errors) + + +def validate_system(original: str, evolved: str) -> ValidationResult: + """校验 System Prompt 改写结果。 + + 检查项: 三个冻结区值比较（能力边界 / 输出格式 / 视频树结构）、 + 长度比在 [0.3, 2.0]、代码块闭合。 + + 参数: + original: 改写前的 system.md 全文。 + evolved: 改写后的 system.md 全文。 + + 返回: + ValidationResult 实例。 + """ + errors: list[str] = [] + frozen_sections = ["能力边界", "输出格式", "视频树结构"] + for section_name in frozen_sections: + orig_section = _extract_section(original, section_name) + if orig_section is None: + continue + evol_section = _extract_section(evolved, section_name) + if evol_section is None: + errors.append(f"冻结区 '## {section_name}' 在改写后缺失") + elif orig_section != evol_section: + errors.append(f"冻结区 '## {section_name}' 在改写后被修改") + errors.extend(_check_length(original, evolved)) + errors.extend(_check_code_blocks(evolved)) + return ValidationResult(passed=len(errors) == 0, errors=errors) + + +def validate_tool( + orig_extract: str, + evol_extract: str, + orig_verify: str, + evol_verify: str, +) -> ValidationResult: + """校验 Tool Prompt 改写结果。 + + 检查项: 输出格式 section 保留（per file）、长度比在 [0.3, 2.0]。 + 与 skill / system 不同，**不检查代码块闭合**。 + + 参数: + orig_extract: 改写前的 extract prompt。 + evol_extract: 改写后的 extract prompt。 + orig_verify: 改写前的 verify prompt。 + evol_verify: 改写后的 verify prompt。 + + 返回: + ValidationResult 实例。 + """ + errors: list[str] = [] + for label, orig, evol in [ + ("extract", orig_extract, evol_extract), + ("verify", orig_verify, evol_verify), + ]: + orig_fmt = _extract_section(orig, "输出格式") + if orig_fmt is not None: + evol_fmt = _extract_section(evol, "输出格式") + if evol_fmt is None: + errors.append(f"{label}: 冻结区 '## 输出格式' 在改写后缺失") + elif orig_fmt != evol_fmt: + errors.append(f"{label}: 冻结区 '## 输出格式' 在改写后被修改") + errors.extend(_check_length(orig, evol)) + return ValidationResult(passed=len(errors) == 0, errors=errors) + + +# ========================================================================= +# D. 纯数学 +# ========================================================================= + + +def edit_budget_at(global_step: int, total_steps: int, start: int, end: int) -> int: + """按 global_step 线性退火的 per-target 编辑预算。 + + 借鉴 SkillOpt LinearScheduler：在 [0, total_steps] 上把预算从 start + 线性退火到 end。total_steps<=1 直接返回 start（避免单步取到最小值）。 + round 用 Python banker's rounding，max(end, ...) 兜底硬下限。 + + 参数: + global_step: 当前全局步（0-indexed）。 + total_steps: 退火地平线（step 数），即分母。 + start: 退火起点（需 >= end）。 + end: 退火终点（亦为硬下限）。 + + 返回: + 当步 per-target 最大 edit 条数。 + + 异常: + AssertionError: start < end。 + """ + assert start >= end, ( + f"edit_budget_at 要求 start >= end，实际 start={start}, end={end}" + ) + if total_steps <= 1: + return start + t = min(global_step, total_steps) / total_steps + return max(end, round(start + (end - start) * t)) + + +# ========================================================================= +# E. JSON 解析（进化版本，不同于 metrics 版） +# ========================================================================= + + +def _parse_llm_json(raw: str) -> dict | None: + """从 LLM 响应中解析 JSON。 + + 仅两种策略：(1) 提取 ```json 代码块；(2) 直接 json.loads。 + 不做 outermost braces 推断、不用 json_repair。失败返回 None。 + + 参数: + raw: LLM 原始输出文本。 + + 返回: + 解析后的字典；失败或结果非 dict 时返回 None。 + """ + text = raw.strip() + # 策略 1：提取 ```json ... ``` 代码块 + code_block = re.search(r"```json\s*\n(.*?)```", text, re.DOTALL) + if code_block: + text = code_block.group(1).strip() + # 策略 2：直接解析 + try: + result = json.loads(text) + if isinstance(result, dict): + return result + return None + except (json.JSONDecodeError, ValueError): + return None + + +# ========================================================================= +# F. rank_and_clip（async） +# ========================================================================= + + +def _select_top_edits( + indices: list[Any], + edits: list[dict[str, Any]], + max_edits: int, +) -> list[dict[str, Any]]: + """按 rank LLM 给出的优先级索引筛选 edits。 + + 依次保留首个 max_edits 条合法、不重复、在范围内的索引对应 edit。 + 用 type(idx) is int（非 isinstance）以排除 bool。 + + 参数: + indices: rank LLM 返回的 0-based 优先级索引。 + edits: 候选 edit 列表。 + max_edits: 最多保留条数。 + + 返回: + 按优先级顺序保留的 edit 列表。 + """ + selected: list[dict[str, Any]] = [] + seen: set[int] = set() + for idx in indices: + if type(idx) is int and 0 <= idx < len(edits) and idx not in seen: + selected.append(edits[idx]) + seen.add(idx) + if len(selected) >= max_edits: + break + return selected + + +async def _request_rank_indices( + llm: LLMProvider, + prompts: str, + original: str, + edits: list[dict[str, Any]], + max_edits: int, + label: str, +) -> list[int]: + """调 rank LLM 取重要性降序的索引列表。 + + 守 P5：响应无法解析或 selected_indices 非列表时直接 raise ValueError。 + + 参数: + llm: LLM 调用端口。 + prompts: evolve_rank 模板内容。 + original: 当前 prompt 全文（排序上下文）。 + edits: 候选 edit 列表。 + max_edits: 本轮预算上限。 + label: 目标标签（仅用于报错信息）。 + + 返回: + rank LLM 返回的原始索引列表（尚未去重/越界过滤）。 + + 异常: + ValueError: 响应无 selected_indices，或其值非列表。 + """ + edits_desc = "\n".join( + f"[{i}] op={e.get('op')} support_count={e.get('support_count', 0)} " + f"target={str(e.get('target', ''))[:60]!r} " + f"content={str(e.get('content', ''))[:60]!r}" + for i, e in enumerate(edits) + ) + user_msg = ( + f"## 当前文件\n\n{original}\n\n" + f"## 候选 edits（{len(edits)} 条，预算 {max_edits} 条）\n\n{edits_desc}\n\n" + f"请选出最重要的 {max_edits} 条，返回其 0-based 索引（重要性降序）。" + ) + response = await llm.chat( + [ + {"role": "system", "content": prompts}, + {"role": "user", "content": user_msg}, + ] + ) + parsed = _parse_llm_json(response.content) + if not parsed or "selected_indices" not in parsed: + raise ValueError(f"{label} rank LLM 未返回 selected_indices，拒绝静默截断") + indices = parsed["selected_indices"] + if not isinstance(indices, list): + raise ValueError(f"{label} rank LLM selected_indices 非列表") + return indices + + +async def rank_and_clip( + llm: LLMProvider, + original_content: str, + edits: list[dict[str, Any]], + max_edits: int, + label: str, + *, + rank_prompt: str = "", +) -> tuple[list[dict[str, Any]], dict[str, Any]]: + """超预算时调 rank LLM 排序取 top-L；未超则原样返回。 + + 三级降级：LLM rank → _select_top_edits → empty → fallback to edits[:max_edits]。 + 对 rank LLM 的输出波动一律优雅降级而非中止。 + + 参数: + llm: LLM 调用端口。 + original_content: 当前 prompt 全文（排序上下文）。 + edits: 候选 edit 列表。 + max_edits: 本轮预算上限。 + label: 目标标签（skill/system/tool，仅用于日志）。 + rank_prompt: evolve_rank 模板内容。 + + 返回: + (裁剪后 edits, {"triggered": bool, "clipped": int})。 + """ + if len(edits) <= max_edits: + return edits, {"triggered": False, "clipped": 0} + + try: + indices = await _request_rank_indices( + llm, rank_prompt, original_content, edits, max_edits, label + ) + except Exception as exc: + logger.warning( + "{} rank LLM 排序不可用（{}）；退化为按原序取前 {} 条", + label, + exc, + max_edits, + ) + indices = [] + + selected = _select_top_edits(indices, edits, max_edits) + if not selected: + selected = edits[:max_edits] + logger.warning("{} rank 有效索引为 0；退化为按原序取前 {} 条", label, max_edits) + elif len(selected) < max_edits: + logger.warning( + "{} rank 仅得 {} 条有效(<预算 {})；按更保守的条数应用", + label, + len(selected), + max_edits, + ) + logger.info("{} edits 超预算裁剪 {}->{}", label, len(edits), len(selected)) + return selected, {"triggered": True, "clipped": len(edits) - len(selected)} + + +# ========================================================================= +# G. resolve_skill_file +# ========================================================================= + + +def resolve_skill_file(skill_store: SkillStore, task_type: str) -> str: + """按运行时规则解析 task_type 对应的 skill 文件名。 + + 转换规则：小写 + 空格替换为短横线 + .md 后缀。 + 若 store 中不存在匹配文件，退化到 default-strategy.md。 + + 参数: + skill_store: 版本化技能读取端口。 + task_type: 题目任务类型（如 "Action Reasoning"）。 + + 返回: + 匹配的 skill 文件名。 + """ + file_name = f"{task_type.lower().replace(' ', '-')}.md" + available = skill_store.list_skill_files() + if file_name in available: + return file_name + return "default-strategy.md" + + +# ========================================================================= +# H. 格式化辅助 +# ========================================================================= + + +def _format_case_samples(cases: list[Any]) -> str: + """将 CaseSample 列表格式化为 LLM 可读文本。 + + 对 trace 中的 tool_output 截断到 500 字符。 + + 参数: + cases: CaseSample 实例列表（也兼容 dict）。 + + 返回: + 格式化后的多行文本。 + """ + lines: list[str] = [] + for case in cases: + if not isinstance(case, dict): + case = asdict(case) + lines.append(f"### {case.get('question_id', 'unknown')}") + lines.append(f"- question: {case.get('question', '')}") + options = case.get("options", []) + if options: + lines.append(f"- options: {json.dumps(options, ensure_ascii=False)}") + lines.append(f"- answer: {case.get('answer', '')}") + lines.append(f"- prediction: {case.get('prediction', '')}") + lines.append(f"- error_type: {case.get('error_type', '')}") + lines.append(f"- selection_reason: {case.get('selection_reason', '')}") + trace = case.get("trace", []) + if trace: + lines.append("- trace:") + for step in trace: + output_text = str(step.get("tool_output", "")) + if len(output_text) > 500: + output_text = output_text[:500] + "..." + lines.append( + f" - step {step.get('step', '?')}: " + f"tool={step.get('tool_name', '')} " + f"args={json.dumps(step.get('tool_args', {}), ensure_ascii=False)} " + f"output={output_text}" + ) + lines.append("") + return "\n".join(lines) + + +def _format_spans(spans: list[dict[str, Any]]) -> str: + """将工具 span 字典列表格式化为 LLM 可读文本。 + + 对 tool_output 截断到 500 字符。 + + 参数: + spans: span 字典列表，每个包含 step / tool_name / tool_args 等字段。 + + 返回: + 格式化后的多行文本。 + """ + lines: list[str] = [] + for span in spans: + lines.append(f"### step {span.get('step', '?')}") + lines.append(f"- tool_name: {span.get('tool_name', '')}") + lines.append( + f"- tool_args: {json.dumps(span.get('tool_args', {}), ensure_ascii=False)}" + ) + output_text = str(span.get("tool_output", "")) + if len(output_text) > 500: + output_text = output_text[:500] + "..." + lines.append(f"- tool_output: {output_text}") + lines.append( + f"- extraction_completeness: {span.get('extraction_completeness', '')}" + ) + lines.append(f"- hallucination_rate: {span.get('hallucination_rate', '')}") + missed = span.get("missed_info_tags", []) + if missed: + lines.append( + f"- missed_info_tags: {json.dumps(missed, ensure_ascii=False)}" + ) + hall_tags = span.get("hallucination_tags", []) + if hall_tags: + lines.append( + f"- hallucination_tags: {json.dumps(hall_tags, ensure_ascii=False)}" + ) + lines.append("") + return "\n".join(lines) + + +def _format_rejected_edits(rejected: list[RejectedEdit]) -> str: + """将已验证无效的改法列表格式化为 LLM 可读文本。 + + gate 证据格式：W=... L=... E={:.2f} delta_hat={:+.3f}。 + + 参数: + rejected: RejectedEdit 实例列表。 + + 返回: + 格式化后的多行文本。 + """ + lines: list[str] = [] + for edit in rejected: + lines.append(f"### {edit.target_file} | delta {edit.delta:+.2f}") + lines.append(f"- 已验证无效的改法: {edit.change_summary}") + if edit.gate_e_value is not None: + lines.append( + f"- 已验证无效: W={edit.gate_w} L={edit.gate_l} " + f"E={edit.gate_e_value:.2f} δ̂={edit.gate_delta_shrunk:+.3f}" + ) + lines.append("") + return "\n".join(lines) diff --git a/tests/unit/test_evolve.py b/tests/unit/test_evolve.py new file mode 100644 index 0000000..2dbd95b --- /dev/null +++ b/tests/unit/test_evolve.py @@ -0,0 +1,659 @@ +"""core/evolution/evolve.py 单元测试。 + +覆盖验证函数、编辑预算退火、resolve_skill_file、内部辅助函数、 +受保护区构建、JSON 解析、rank_and_clip、格式化工具。 +""" + +from __future__ import annotations + +import json +from unittest.mock import AsyncMock + +import pytest + +from core.evolution.evolve import ( + _check_code_blocks, + _check_length, + _extract_section, + _format_case_samples, + _format_rejected_edits, + _format_spans, + _parse_frontmatter, + _parse_llm_json, + _select_top_edits, + _skill_protected_spans, + _strip_appendix_region, + _strip_momentum_region, + _strip_protected_regions, + _system_protected_spans, + _tool_protected_spans, + edit_budget_at, + rank_and_clip, + resolve_skill_file, + validate_skill, + validate_system, + validate_tool, +) +from core.evolution.patch import ( + APPENDIX_END, + APPENDIX_START, + MOMENTUM_END, + MOMENTUM_START, +) +from core.evolution.types import RejectedEdit + +# ========================================================================= +# A. 内部辅助函数 +# ========================================================================= + + +class TestParseFrontmatter: + """_parse_frontmatter 测试。""" + + def test_valid_frontmatter(self) -> None: + text = "---\nname: test\ndescription: d\n---\nbody" + result = _parse_frontmatter(text) + assert result == {"name": "test", "description": "d"} + + def test_no_frontmatter(self) -> None: + assert _parse_frontmatter("no frontmatter here") is None + + def test_invalid_yaml(self) -> None: + text = "---\n: : invalid\n---\nbody" + assert _parse_frontmatter(text) is None + + def test_empty_frontmatter(self) -> None: + text = "---\n\n---\nbody" + result = _parse_frontmatter(text) + assert result is None # yaml.safe_load("") returns None + + +class TestStripAppendixRegion: + """_strip_appendix_region 测试。""" + + def test_no_markers(self) -> None: + text = "hello world" + assert _strip_appendix_region(text) == text + + def test_with_markers(self) -> None: + text = f"before\n{APPENDIX_START}\nappendix stuff\n{APPENDIX_END}\nafter" + result = _strip_appendix_region(text) + assert "appendix stuff" not in result + assert "before" in result + assert "after" in result + + def test_only_start_marker(self) -> None: + text = f"before\n{APPENDIX_START}\nno end marker" + assert _strip_appendix_region(text) == text + + def test_only_end_marker(self) -> None: + text = f"before\n{APPENDIX_END}\nno start marker" + assert _strip_appendix_region(text) == text + + +class TestStripMomentumRegion: + """_strip_momentum_region 测试。""" + + def test_no_markers(self) -> None: + text = "hello world" + assert _strip_momentum_region(text) == text + + def test_with_markers(self) -> None: + text = f"before\n{MOMENTUM_START}\nmomentum stuff\n{MOMENTUM_END}\nafter" + result = _strip_momentum_region(text) + assert "momentum stuff" not in result + assert "before" in result + assert "after" in result + + def test_damaged_markers_raise(self) -> None: + text = f"before\n{MOMENTUM_START}\nno end marker" + with pytest.raises(ValueError, match="momentum marker 损坏"): + _strip_momentum_region(text) + + +class TestStripProtectedRegions: + """_strip_protected_regions 测试（先 appendix 后 momentum）。""" + + def test_both_regions(self) -> None: + text = ( + f"body\n" + f"{APPENDIX_START}\nappendix\n{APPENDIX_END}\n" + f"{MOMENTUM_START}\nmomentum\n{MOMENTUM_END}\n" + f"tail" + ) + result = _strip_protected_regions(text) + assert "appendix" not in result + assert "momentum" not in result + assert "body" in result + assert "tail" in result + + +class TestCheckLength: + """_check_length 测试。""" + + def test_normal_ratio(self) -> None: + orig = "x" * 100 + evol = "x" * 120 + assert _check_length(orig, evol) == [] + + def test_too_long(self) -> None: + orig = "x" * 100 + evol = "x" * 300 + errors = _check_length(orig, evol) + assert len(errors) == 1 + assert "超限" in errors[0] + + def test_too_short(self) -> None: + orig = "x" * 100 + evol = "x" * 10 + errors = _check_length(orig, evol) + assert len(errors) == 1 + assert "不足" in errors[0] + + def test_orig_empty(self) -> None: + assert _check_length("", "something") == [] + + +class TestCheckCodeBlocks: + """_check_code_blocks 测试。""" + + def test_even_count(self) -> None: + text = "```python\ncode\n```" + assert _check_code_blocks(text) == [] + + def test_odd_count(self) -> None: + text = "```python\ncode" + errors = _check_code_blocks(text) + assert len(errors) == 1 + assert "未闭合" in errors[0] + + def test_no_blocks(self) -> None: + assert _check_code_blocks("no code blocks") == [] + + +class TestExtractSection: + """_extract_section 测试。""" + + def test_found(self) -> None: + text = "intro\n## 能力边界\nfrozen content\n## other\nmore" + result = _extract_section(text, "能力边界") + assert result is not None + assert "frozen content" in result + assert "## 能力边界" in result + + def test_not_found(self) -> None: + text = "intro\n## other\nmore" + assert _extract_section(text, "不存在") is None + + def test_last_section(self) -> None: + text = "intro\n## 最后段\ncontent at end" + result = _extract_section(text, "最后段") + assert result is not None + assert "content at end" in result + + +# ========================================================================= +# B. 受保护区构建 +# ========================================================================= + + +class TestSkillProtectedSpans: + """_skill_protected_spans 测试。""" + + def test_with_frontmatter(self) -> None: + text = "---\nname: test\n---\nbody" + spans = _skill_protected_spans(text) + assert any("---" in s for s in spans) + + def test_with_appendix(self) -> None: + text = f"body\n{APPENDIX_START}\nnotes\n{APPENDIX_END}" + spans = _skill_protected_spans(text) + assert any(APPENDIX_START in s for s in spans) + + def test_with_momentum(self) -> None: + text = f"body\n{MOMENTUM_START}\nmomentum\n{MOMENTUM_END}" + spans = _skill_protected_spans(text) + assert any(MOMENTUM_START in s for s in spans) + + def test_empty(self) -> None: + assert _skill_protected_spans("plain text") == [] + + +class TestSystemProtectedSpans: + """_system_protected_spans 测试。""" + + def test_frozen_sections(self) -> None: + text = "intro\n## 能力边界\ncontent1\n## 输出格式\ncontent2\n## 视频树结构\ncontent3\n## other\nmore" + spans = _system_protected_spans(text) + assert len(spans) == 3 + + def test_with_appendix(self) -> None: + text = f"body\n{APPENDIX_START}\nnotes\n{APPENDIX_END}" + spans = _system_protected_spans(text) + assert len(spans) == 1 + + +class TestToolProtectedSpans: + """_tool_protected_spans 测试。""" + + def test_output_format_section(self) -> None: + text = "intro\n## 输出格式\nformat\n## other\nmore" + spans = _tool_protected_spans(text) + assert any("输出格式" in s for s in spans) + + def test_no_output_format(self) -> None: + text = "intro\n## other\nmore" + spans = _tool_protected_spans(text) + assert len(spans) == 0 + + +# ========================================================================= +# C. 验证函数 +# ========================================================================= + + +class TestValidateSkill: + """validate_skill 测试。""" + + def test_identical_passes(self) -> None: + content = "---\nname: test\ndescription: d\ntask_type: t\n---\nbody" + assert validate_skill(content, content).passed + + def test_changed_frontmatter_fails(self) -> None: + orig = "---\nname: a\ndescription: d\ntask_type: t\n---\nbody" + evol = "---\nname: b\ndescription: d\ntask_type: t\n---\nbody" + result = validate_skill(orig, evol) + assert not result.passed + assert any("name" in e for e in result.errors) + + def test_length_ratio_too_short_fails(self) -> None: + orig = "---\nname: a\ndescription: d\ntask_type: t\n---\n" + "x" * 1000 + evol = "---\nname: a\ndescription: d\ntask_type: t\n---\nshort" + result = validate_skill(orig, evol) + assert not result.passed + assert any("不足" in e for e in result.errors) + + def test_length_ratio_too_long_fails(self) -> None: + orig = "---\nname: a\ndescription: d\ntask_type: t\n---\nshort" + evol = "---\nname: a\ndescription: d\ntask_type: t\n---\n" + "x" * 1000 + result = validate_skill(orig, evol) + assert not result.passed + assert any("超限" in e for e in result.errors) + + def test_unclosed_code_block_fails(self) -> None: + orig = "---\nname: a\ndescription: d\ntask_type: t\n---\nbody" + evol = "---\nname: a\ndescription: d\ntask_type: t\n---\nbody\n```" + result = validate_skill(orig, evol) + assert not result.passed + assert any("未闭合" in e for e in result.errors) + + def test_missing_original_frontmatter(self) -> None: + result = validate_skill("no frontmatter", "no frontmatter") + assert not result.passed + assert any("原文缺少" in e for e in result.errors) + + def test_missing_evolved_frontmatter(self) -> None: + orig = "---\nname: a\ndescription: d\ntask_type: t\n---\nbody" + result = validate_skill(orig, "no frontmatter body") + assert not result.passed + + +class TestValidateSystem: + """validate_system 测试。""" + + def test_identical_passes(self) -> None: + content = "intro\n## 能力边界\nfrozen\n## 输出格式\nfrozen2\n## other\nbody" + assert validate_system(content, content).passed + + def test_changed_frozen_section_fails(self) -> None: + orig = "intro\n## 能力边界\noriginal\n## other\nbody" + evol = "intro\n## 能力边界\nchanged\n## other\nbody" + result = validate_system(orig, evol) + assert not result.passed + assert any("能力边界" in e for e in result.errors) + + def test_missing_frozen_section_fails(self) -> None: + orig = "intro\n## 能力边界\nfrozen\n## other\nbody" + evol = "intro\n## other\nbody that is similar length content padding" + result = validate_system(orig, evol) + assert not result.passed + assert any("缺失" in e for e in result.errors) + + def test_no_frozen_sections_passes(self) -> None: + content = "intro\n## other section\nbody content" + assert validate_system(content, content).passed + + def test_code_block_check(self) -> None: + content = "## 能力边界\nfrozen\n## other\nbody\n```unclosed" + result = validate_system(content, content) + assert not result.passed + + +class TestValidateTool: + """validate_tool 测试。""" + + def test_identical_passes(self) -> None: + extract = "## 输出格式\nfixed\n## other\nbody" + verify = "## 输出格式\nfixed2\n## other\nbody2" + assert validate_tool(extract, extract, verify, verify).passed + + def test_no_code_block_check(self) -> None: + """validate_tool 不检查代码块闭合（与 skill/system 不同）。""" + extract = "## 输出格式\nfixed\n```\nunclosed" + assert validate_tool(extract, extract, "v", "v").passed + + def test_changed_output_format_fails(self) -> None: + orig = "## 输出格式\noriginal format\n## other\nbody" + evol = "## 输出格式\nchanged format\n## other\nbody" + result = validate_tool(orig, evol, "v", "v") + assert not result.passed + assert any("输出格式" in e for e in result.errors) + + def test_verify_output_format_checked_too(self) -> None: + extract = "## 输出格式\nfixed\n## other\nbody" + orig_verify = "## 输出格式\nfixed_v\n## other\nbody_v" + evol_verify = "## 输出格式\nchanged_v\n## other\nbody_v" + result = validate_tool(extract, extract, orig_verify, evol_verify) + assert not result.passed + + +# ========================================================================= +# D. 纯数学 +# ========================================================================= + + +class TestEditBudget: + """edit_budget_at 测试。""" + + def test_start_at_zero(self) -> None: + assert edit_budget_at(0, 100, 5, 2) == 5 + + def test_end_at_total(self) -> None: + assert edit_budget_at(100, 100, 5, 2) == 2 + + def test_total_steps_one(self) -> None: + assert edit_budget_at(0, 1, 5, 2) == 5 + + def test_start_less_than_end_asserts(self) -> None: + with pytest.raises(AssertionError): + edit_budget_at(0, 100, 2, 5) + + def test_mid_step(self) -> None: + result = edit_budget_at(50, 100, 5, 2) + assert 2 <= result <= 5 + + def test_beyond_total_clamped(self) -> None: + """global_step 超过 total_steps 时被钳住在 end。""" + assert edit_budget_at(200, 100, 5, 2) == 2 + + def test_equal_start_end(self) -> None: + assert edit_budget_at(50, 100, 3, 3) == 3 + + def test_total_steps_zero(self) -> None: + """total_steps <= 1 直接返回 start。""" + assert edit_budget_at(0, 0, 5, 2) == 5 + + +# ========================================================================= +# E. JSON 解析 +# ========================================================================= + + +class TestParseLlmJson: + """_parse_llm_json 测试。""" + + def test_plain_json(self) -> None: + raw = '{"key": "value"}' + assert _parse_llm_json(raw) == {"key": "value"} + + def test_fenced_json(self) -> None: + raw = 'text before\n```json\n{"key": "value"}\n```\ntext after' + assert _parse_llm_json(raw) == {"key": "value"} + + def test_non_dict_returns_none(self) -> None: + raw = "[1, 2, 3]" + assert _parse_llm_json(raw) is None + + def test_invalid_json_returns_none(self) -> None: + raw = "not json at all" + assert _parse_llm_json(raw) is None + + def test_empty_string(self) -> None: + assert _parse_llm_json("") is None + + +# ========================================================================= +# F. rank_and_clip +# ========================================================================= + + +class TestSelectTopEdits: + """_select_top_edits 测试。""" + + def test_valid_indices(self) -> None: + edits = [{"op": "a"}, {"op": "b"}, {"op": "c"}] + result = _select_top_edits([2, 0], edits, 2) + assert result == [{"op": "c"}, {"op": "a"}] + + def test_dedup(self) -> None: + edits = [{"op": "a"}, {"op": "b"}] + result = _select_top_edits([0, 0, 1], edits, 3) + assert len(result) == 2 + + def test_out_of_bounds_skipped(self) -> None: + edits = [{"op": "a"}, {"op": "b"}] + result = _select_top_edits([5, 0, -1], edits, 3) + assert result == [{"op": "a"}] + + def test_bool_excluded(self) -> None: + """type(True) is int 返回 True 但 type(idx) is int 排除 bool。""" + edits = [{"op": "a"}, {"op": "b"}] + result = _select_top_edits([True, False, 0], edits, 3) + assert result == [{"op": "a"}] + + def test_max_edits_limit(self) -> None: + edits = [{"op": "a"}, {"op": "b"}, {"op": "c"}] + result = _select_top_edits([0, 1, 2], edits, 2) + assert len(result) == 2 + + +class TestRankAndClip: + """rank_and_clip 测试。""" + + @pytest.mark.asyncio + async def test_within_budget_passthrough(self) -> None: + """edits <= max_edits 时原样返回。""" + llm = AsyncMock() + edits = [{"op": "a"}, {"op": "b"}] + result, info = await rank_and_clip(llm, "content", edits, 5, "test") + assert result == edits + assert info["triggered"] is False + llm.chat.assert_not_called() + + @pytest.mark.asyncio + async def test_llm_rank_success(self) -> None: + """LLM 成功返回索引时按优先级裁剪。""" + llm = AsyncMock() + response = AsyncMock() + response.content = json.dumps({"selected_indices": [2, 0]}) + llm.chat.return_value = response + + edits = [{"op": "a"}, {"op": "b"}, {"op": "c"}] + result, info = await rank_and_clip(llm, "content", edits, 2, "test") + assert len(result) == 2 + assert result[0] == {"op": "c"} + assert info["triggered"] is True + + @pytest.mark.asyncio + async def test_llm_failure_fallback(self) -> None: + """LLM 失败时退化为按原序取前 max_edits 条。""" + llm = AsyncMock() + llm.chat.side_effect = Exception("LLM down") + + edits = [{"op": "a"}, {"op": "b"}, {"op": "c"}] + result, info = await rank_and_clip(llm, "content", edits, 2, "test") + assert len(result) == 2 + assert result == [{"op": "a"}, {"op": "b"}] + assert info["triggered"] is True + + +# ========================================================================= +# G. resolve_skill_file +# ========================================================================= + + +class TestResolveSkillFile: + """resolve_skill_file 测试。""" + + def test_direct_match(self) -> None: + class FakeStore: + def list_skill_files(self) -> list[str]: + return ["action-reasoning.md", "default-strategy.md"] + + def read_skill(self, f: str) -> str: + return "" + + assert resolve_skill_file(FakeStore(), "Action Reasoning") == "action-reasoning.md" + + def test_fallback_to_default(self) -> None: + class FakeStore: + def list_skill_files(self) -> list[str]: + return ["default-strategy.md"] + + def read_skill(self, f: str) -> str: + return "" + + assert resolve_skill_file(FakeStore(), "Unknown Type") == "default-strategy.md" + + def test_case_insensitive(self) -> None: + class FakeStore: + def list_skill_files(self) -> list[str]: + return ["temporal-reasoning.md"] + + def read_skill(self, f: str) -> str: + return "" + + assert resolve_skill_file(FakeStore(), "Temporal Reasoning") == "temporal-reasoning.md" + + +# ========================================================================= +# H. 格式化辅助 +# ========================================================================= + + +class TestFormatCaseSamples: + """_format_case_samples 测试。""" + + def test_basic_format(self) -> None: + cases = [ + { + "question_id": "q1", + "question": "What?", + "options": ["A", "B"], + "answer": "A", + "prediction": "B", + "error_type": "wrong", + "selection_reason": "test", + "trace": [], + } + ] + result = _format_case_samples(cases) + assert "q1" in result + assert "What?" in result + + def test_trace_truncation(self) -> None: + cases = [ + { + "question_id": "q1", + "question": "Q", + "options": [], + "answer": "A", + "prediction": "B", + "error_type": "e", + "selection_reason": "s", + "trace": [ + { + "step": 1, + "tool_name": "t", + "tool_args": {}, + "tool_output": "x" * 600, + } + ], + } + ] + result = _format_case_samples(cases) + assert "..." in result + + +class TestFormatSpans: + """_format_spans 测试。""" + + def test_basic_format(self) -> None: + spans = [ + { + "step": 1, + "tool_name": "extract", + "tool_args": {"query": "test"}, + "tool_output": "result", + "extraction_completeness": 0.9, + "hallucination_rate": 0.1, + } + ] + result = _format_spans(spans) + assert "extract" in result + assert "0.9" in result + + def test_output_truncation(self) -> None: + spans = [ + { + "step": 1, + "tool_name": "t", + "tool_args": {}, + "tool_output": "x" * 600, + "extraction_completeness": 0.5, + "hallucination_rate": 0.0, + } + ] + result = _format_spans(spans) + assert "..." in result + + +class TestFormatRejectedEdits: + """_format_rejected_edits 测试。""" + + def test_with_gate_evidence(self) -> None: + edits = [ + RejectedEdit( + target_file="skill.md", + target_type="skill", + change_summary="changed X", + delta=-0.05, + source_version="v2", + epoch=3, + gate_w=5, + gate_l=8, + gate_e_value=0.42, + gate_delta_shrunk=-0.123, + ) + ] + result = _format_rejected_edits(edits) + assert "W=5" in result + assert "L=8" in result + assert "E=0.42" in result + assert "δ̂=-0.123" in result + + def test_without_gate_evidence(self) -> None: + edits = [ + RejectedEdit( + target_file="skill.md", + target_type="skill", + change_summary="changed X", + delta=-0.05, + source_version="v2", + epoch=3, + ) + ] + result = _format_rejected_edits(edits) + assert "skill.md" in result + assert "changed X" in result + assert "W=" not in result