diff --git a/core/evolution/patch.py b/core/evolution/patch.py new file mode 100644 index 0000000..d4cef82 --- /dev/null +++ b/core/evolution/patch.py @@ -0,0 +1,427 @@ +"""定点补丁引擎:把进化输出的离散 edit 逐条应用到文本,逐条出状态报告。 + +借鉴 SkillOpt skill.py 的 apply 语义;守 P5:找不到锚点不静默乱改、不裸 except。 +冻结区按全文坐标区间判定;append/退化追加插到最早冻结区之前(无则 EOF)。 +""" + +from __future__ import annotations + +from loguru import logger + +APPENDIX_START = "" +APPENDIX_END = "" +APPENDIX_MAX_CHARS = 2000 # appendix 区软上限(守设计「长度上限+warning,不做去重」) + +MOMENTUM_START = "" +MOMENTUM_END = "" +MOMENTUM_MAX_CHARS = 2000 # momentum 区软上限(与 appendix 一致:超限 warning 不截断) +MOMENTUM_HEADING = ( + "## 动量指导(每轮重写,勿手改)" # replace_momentum 写入的固定标题行 +) + + +def momentum_region_bounds(text: str) -> tuple[int, int] | None: + """定位 momentum 受保护区的字符区间,并对损坏态显式报错(P5)。 + + momentum marker 由 replace_momentum 在 epoch 末反复重写,guidance 又来自 LLM + 外部输入,因此 marker 可能出现损坏态。本函数是 momentum 路径的唯一边界判定入口, + 把配对校验集中在一处: + + - START 与 END 各恰好出现一次且 START 在 END 之前 → 返回 (start_idx, end_idx), + end_idx 指向 END marker 结束位置(即 content[start:end] 含完整两 marker)。 + - 两 marker 都不出现 → 返回 None(合法的"无区"态,调用方据此新建)。 + - 其余皆为损坏态(仅一个 marker、END 在 START 前、任一 marker 重复)→ raise + ValueError,拒绝静默新建/跳过,要求人工修复。 + + 参数: + text: 待检测的文本(skill 全文)。 + 返回: + (start_idx, end_idx) 表示区间,或 None 表示无 momentum 区。 + 异常: + ValueError: momentum marker 损坏/不配对。 + """ + start_count = text.count(MOMENTUM_START) + end_count = text.count(MOMENTUM_END) + if start_count == 0 and end_count == 0: + return None + if start_count != 1 or end_count != 1: + raise ValueError( + f"momentum marker 损坏/不配对:MOMENTUM_START 出现 {start_count} 次、" + f"MOMENTUM_END 出现 {end_count} 次(各须恰好 1 次),需人工修复" + ) + start_idx = text.index(MOMENTUM_START) + end_idx = text.index(MOMENTUM_END) + len(MOMENTUM_END) + if start_idx >= text.index(MOMENTUM_END): + raise ValueError( + "momentum marker 损坏/不配对:MOMENTUM_END 出现在 MOMENTUM_START 之前,需人工修复" + ) + return start_idx, end_idx + + +def momentum_inner(content: str) -> str: + """返回 momentum 受保护区的内层文本(去掉两 marker),无区返回空串。 + + 与 _momentum_span(含 marker 的整段)的区别:本函数只取两 marker 之间的内层正文, + 供 run_slow_momentum 的 prev_guidance 使用。prev_guidance 在 LLM 解析失败时会被 + run_slow_momentum 原样返回、再喂给 replace_momentum;replace_momentum 禁止 guidance + 含 marker 字面量,故 prev_guidance 必须是无 marker 的内层文本,否则一旦解析回退即 + 在 replace_momentum 抛 ValueError。 + + 边界判定与配对校验统一委托 momentum_region_bounds:marker 损坏/不配对时由其 raise + ValueError,本函数不把损坏态静默当作"无区"。 + + 参数: + content: skill 全文。 + 返回: + momentum 区两 marker 之间的内层文本(已 strip);无区返回空串。 + 异常: + ValueError: momentum marker 损坏/不配对。 + """ + bounds = momentum_region_bounds(content) + if bounds is None: + return "" + start, end = bounds + inner = content[start + len(MOMENTUM_START) : end - len(MOMENTUM_END)].strip() + # 去掉 replace_momentum 写入的固定标题行,只回传纯指导文本,使其等价于上一轮 + # 传给 replace_momentum 的 guidance(解析回退时原样回传不会引入重复标题)。 + if inner.startswith(MOMENTUM_HEADING): + inner = inner[len(MOMENTUM_HEADING) :].lstrip("\n") + return inner.strip() + + +def append_to_appendix(content: str, notes: list[str]) -> str: + """把 LAPSE 提醒追加到文件尾的 appendix 受保护区;区不存在则创建。 + + 护栏:appendix 区超过 APPENDIX_MAX_CHARS 时 logger.warning(不静默截断, + 提示人工压缩;不做自动去重——YAGNI,见设计)。 + + 参数: + content: 原文。 + notes: 待追加的提醒文本列表。 + 返回: + 含 appendix 区的新文本。 + """ + if not notes: + return content + bullet = "\n".join(f"- {n.strip()}" for n in notes if n.strip()) + if not bullet: + return content + if APPENDIX_START in content and APPENDIX_END in content: + head, rest = content.split(APPENDIX_START, 1) + inner, tail = rest.split(APPENDIX_END, 1) + new_inner = f"{inner.rstrip()}\n{bullet}" + out = f"{head}{APPENDIX_START}{new_inner}\n{APPENDIX_END}{tail}" + else: + new_inner = f"\n## 执行提醒(自动累积,勿手改)\n{bullet}" + out = f"{content.rstrip()}\n\n{APPENDIX_START}{new_inner}\n{APPENDIX_END}\n" + if len(new_inner) > APPENDIX_MAX_CHARS: + logger.warning( + "appendix 区长度 {} 超过上限 {},建议人工压缩", + len(new_inner), + APPENDIX_MAX_CHARS, + ) + return out + + +def appendix_region_bounds(text: str) -> tuple[int, int] | None: + """定位 appendix 受保护区的字符区间,对损坏态显式报错(P5,对称 momentum_region_bounds)。 + + appendix marker 由 append_to_appendix 维护、consolidation 回写,可能出现损坏态。 + 本函数是 appendix 路径的唯一边界判定入口,把配对校验集中一处: + + - START 与 END 各恰好一次且 START 在 END 之前 → 返回 (start_idx, end_idx), + end_idx 指向 END marker 结束位置(content[start:end] 含完整两 marker)。 + - 两 marker 都不出现 → 返回 None(合法的「无区」态)。 + - 其余(仅一个 marker、END 在 START 前、任一 marker 重复)→ raise ValueError, + 拒绝静默按字符串切片处理而误拼/吞掉区外正文。 + + 参数: + text: 待检测文本(skill 全文)。 + 返回: + (start_idx, end_idx) 表示区间,或 None 表示无 appendix 区。 + 异常: + ValueError: appendix marker 损坏/不配对。 + """ + start_count = text.count(APPENDIX_START) + end_count = text.count(APPENDIX_END) + if start_count == 0 and end_count == 0: + return None + if start_count != 1 or end_count != 1: + raise ValueError( + f"appendix marker 损坏/不配对:APPENDIX_START 出现 {start_count} 次、" + f"APPENDIX_END 出现 {end_count} 次(各须恰好 1 次),需人工修复" + ) + start_idx = text.index(APPENDIX_START) + end_idx = text.index(APPENDIX_END) + len(APPENDIX_END) + if start_idx >= text.index(APPENDIX_END): + raise ValueError( + "appendix marker 损坏/不配对:APPENDIX_END 出现在 APPENDIX_START 之前,需人工修复" + ) + return start_idx, end_idx + + +def extract_appendix_notes(content: str) -> list[str]: + """从 appendix 受保护区解析出 bullet 提醒列表;无区返回空列表。 + + 功能: + 取 appendix 区内每行以 "- " 起头的文本为一条 note(去 "- " 前缀与首尾空白), + 区内标题行(## 执行提醒…)不计。供 consolidation 读取现有 notes。 + 参数: + content: skill 全文。 + 返回: + note 字符串列表;无 appendix 区返回 []。 + 异常: + ValueError: appendix marker 损坏/不配对(经 appendix_region_bounds,不静默切片)。 + 关键实现细节: + 边界判定统一委托 appendix_region_bounds,只取两 marker 之间内层正文逐行解析。 + """ + bounds = appendix_region_bounds(content) + if bounds is None: + return [] + start, end = bounds + inner = content[start + len(APPENDIX_START) : end - len(APPENDIX_END)] + notes: list[str] = [] + for line in inner.splitlines(): + stripped = line.strip() + if stripped.startswith("- "): + note = stripped[2:].strip() + if note: + notes.append(note) + return notes + + +def replace_appendix_notes(content: str, notes: list[str]) -> str: + """用 notes 整体替换 appendix 区内容;notes 为空则删除整个 appendix 区。 + + 功能: + consolidation 回写压缩后 notes 的替换语义(区别于 append_to_appendix 累积): + 区存在则整体覆盖区内 bullet;notes 空则连 marker 一并删除、保留区外正文; + 区不存在且 notes 非空则按 append_to_appendix 格式新建。 + 参数: + content: 原文(可能含 appendix 区)。 + notes: 压缩后的提醒列表;空列表表示删区。 + 返回: + 替换后的全文。 + 异常: + ValueError: appendix marker 损坏/不配对(经 appendix_region_bounds)。 + 关键实现细节: + 边界经 appendix_region_bounds 显式校验,按 (start,end) 切出 head/tail 拼接, + 不做两次独立 split(避免损坏态误拼/吞掉区外正文)。 + """ + bounds = appendix_region_bounds(content) + if bounds is not None: + start, end = bounds + head = content[:start] + tail = content[end:] + if not notes: + return head.rstrip() + ("\n" + tail.lstrip("\n") if tail.strip() else "\n") + bullet = "\n".join(f"- {n.strip()}" for n in notes if n.strip()) + new_inner = f"\n## 执行提醒(自动累积,勿手改)\n{bullet}" + return f"{head}{APPENDIX_START}{new_inner}\n{APPENDIX_END}{tail}" + if not notes: + return content + return append_to_appendix(content, notes) + + +def replace_momentum(content: str, guidance: str) -> str: + """把「动量指导」整体写入文件尾的 momentum 受保护区;区不存在则创建。 + + 与 append_to_appendix 的累积语义不同,momentum 是**替换**语义:慢更新周期每 + epoch 末整体重写一段动量指导,旧指导被完全覆盖(不保留历史)。momentum 区与 + appendix 区独立共存——本函数只触碰 momentum marker,不破坏已有 appendix 区。 + + 护栏:momentum 区超过 MOMENTUM_MAX_CHARS 时 logger.warning(不静默截断,与 + appendix 对齐)。 + + 关键实现细节: + - 替换非追加:区已存在时用 guidance 整体覆盖 marker 内 inner,旧动量不残留。 + - 创建位置在文件尾(append_to_appendix 同样在文件尾,但两区 marker 不同, + split 按各自 marker 定位,互不干扰)。 + + 空 guidance 决策:与 appendix 的累积语义不同,momentum 是「每轮整体重写」,空 + guidance 表示「本轮无动量指导」,属合法语义——照常写入(区内仅留标题,旧动量被清空), + 而非返回原文保留旧动量。 + + 参数: + content: 原文(可能已含 appendix 区)。 + guidance: 本轮动量指导全文(整体覆盖旧动量)。 + 返回: + 含 momentum 区的新文本。 + 异常: + ValueError: guidance 含 momentum marker 字面量(外部输入注入),或原文 momentum + marker 损坏/不配对。 + """ + if MOMENTUM_START in guidance or MOMENTUM_END in guidance: + raise ValueError( + "guidance 不得包含 momentum marker 字面量" + f"({MOMENTUM_START} / {MOMENTUM_END}),否则会破坏 marker 配对" + ) + bounds = momentum_region_bounds(content) + new_inner = f"\n## 动量指导(每轮重写,勿手改)\n{guidance.strip()}" + if bounds is not None: + start_idx, end_idx = bounds + head = content[:start_idx] + tail = content[end_idx:] + out = f"{head}{MOMENTUM_START}{new_inner}\n{MOMENTUM_END}{tail}" + else: + out = f"{content.rstrip()}\n\n{MOMENTUM_START}{new_inner}\n{MOMENTUM_END}\n" + if len(new_inner) > MOMENTUM_MAX_CHARS: + logger.warning( + "momentum 区长度 {} 超过上限 {},建议人工压缩", + len(new_inner), + MOMENTUM_MAX_CHARS, + ) + return out + + +def _protected_ranges(content: str, spans: list[str]) -> list[tuple[int, int]]: + """把冻结文本块映射成 content 中的 [start, end) 坐标区间。""" + ranges: list[tuple[int, int]] = [] + for span in spans: + idx = content.find(span) + if idx != -1: + ranges.append((idx, idx + len(span))) + return ranges + + +def _in_ranges(pos: int, ranges: list[tuple[int, int]]) -> bool: + """判断位置 pos 是否落在任意冻结区间内。""" + return any(start <= pos < end for start, end in ranges) + + +def _append_at(content: str, ranges: list[tuple[int, int]]) -> int: + """append/退化追加落点:最早一个 start>0 的冻结区之前;无则文末(头部 frontmatter 不计)。""" + starts = [start for start, _ in ranges if start > 0] + return min(starts) if starts else len(content) + + +def _insert_at(content: str, at: int, payload: str) -> str: + """在 at 位置插入 payload,自动补换行保持段落格式。""" + head, tail = content[:at].rstrip(), content[at:].lstrip("\n") + if tail: + return head + "\n\n" + payload + "\n\n" + tail + return head + "\n\n" + payload + "\n" + + +def _do_append( + content: str, payload: str, ranges: list[tuple[int, int]] +) -> tuple[str, str]: + """执行 append 操作,返回更新后内容与状态字符串。""" + return _insert_at(content, _append_at(content, ranges), payload), "applied_append" + + +def _do_insert_after( + content: str, target: str, payload: str, ranges: list[tuple[int, int]] +) -> tuple[str, str]: + """执行 insert_after 操作,处理退化追加与冻结区跳过。""" + pos = content.find(target) if target else -1 + if pos == -1: + logger.warning("insert_after 锚点缺失,退化为追加 target={}", target[:80]) + return ( + _insert_at(content, _append_at(content, ranges), payload), + "applied_insert_after_fallback", + ) + if _in_ranges(pos, ranges): + logger.warning("insert_after 目标在冻结区,跳过 target={}", target[:80]) + return content, "skipped_protected" + at = pos + len(target) + nl = content.find("\n", at) + at = nl + 1 if nl != -1 else len(content) + return content[:at] + payload + "\n" + content[at:], "applied_insert_after" + + +def _do_replace_delete( + op: str, + content: str, + target: str, + payload: str, + ranges: list[tuple[int, int]], +) -> tuple[str, str]: + """执行 replace 或 delete 操作,返回更新后内容与状态字符串。""" + if not target: + return content, "skipped_missing_target" + pos = content.find(target) + if pos == -1: + logger.warning("{} 锚点缺失,跳过 target={}", op, target[:80]) + return content, "skipped_target_not_found" + if _in_ranges(pos, ranges): + logger.warning("{} 目标在冻结区,跳过 target={}", op, target[:80]) + return content, "skipped_protected" + new_content = content.replace(target, payload if op == "replace" else "", 1) + return new_content, "applied_" + op + + +def _apply_one( + content: str, edit: dict, ranges: list[tuple[int, int]] +) -> tuple[str, dict]: + """应用单条 edit,返回 (更新后内容, 状态报告)。""" + if not isinstance(edit, dict): + return content, { + "op": "", + "target": "", + "content_preview": "", + "status": "error", + "error": f"edit 非 dict: {type(edit).__name__}", + } + op = str(edit.get("op", "")) + target = str(edit.get("target", "") or "") + payload = str(edit.get("content", "") or "").strip() + report = { + "op": op, + "target": target[:200], + "content_preview": payload[:200], + "status": "unknown", + } + + if op == "append": + content, report["status"] = _do_append(content, payload, ranges) + return content, report + + if op == "insert_after": + content, report["status"] = _do_insert_after(content, target, payload, ranges) + return content, report + + if op in ("replace", "delete"): + content, report["status"] = _do_replace_delete( + op, content, target, payload, ranges + ) + return content, report + + logger.warning("未知 op,跳过: {}", op) + report["status"] = "skipped_unknown_op" + return content, report + + +def apply_patch_with_report( + content: str, + edits: list[dict], + protected_spans: list[str] | None = None, +) -> tuple[str, list[dict]]: + """顺序应用 edit 列表,返回 (新内容, 逐条状态报告)。 + + 参数: + content: 原始文本。 + edits: 每条 {op, target, content}。 + protected_spans: 冻结文本块列表;目标落入其坐标区间即跳过,append 插到其前。 + + 返回: + (应用后文本, reports);reports 每条含 op/target/content_preview/status/index。 + """ + spans = protected_spans or [] + reports: list[dict] = [] + for i, edit in enumerate(edits, 1): + try: + ranges = _protected_ranges(content, spans) + content, report = _apply_one(content, edit, ranges) + except (KeyError, TypeError, ValueError, AttributeError) as exc: + report = { + "op": "", + "target": "", + "content_preview": "", + "status": "error", + "error": str(exc), + } + logger.exception("补丁应用异常 index={}", i) + report["index"] = i + reports.append(report) + return content, reports diff --git a/tests/unit/test_patch.py b/tests/unit/test_patch.py new file mode 100644 index 0000000..b6a029d --- /dev/null +++ b/tests/unit/test_patch.py @@ -0,0 +1,391 @@ +"""patch.py 补丁引擎单元测试。 + +覆盖四大区域: +1. TestRegionBounds — appendix/momentum 边界定位与损坏态检测 +2. TestAppendix — 追加/提取/替换 appendix 区 +3. TestMomentum — 替换/提取 momentum 区 +4. TestApplyPatch — apply_patch_with_report 的 4 种 op + 冻结区 + 异常 +""" + +from __future__ import annotations + +import pytest + +from core.evolution.patch import ( + APPENDIX_END, + APPENDIX_MAX_CHARS, + APPENDIX_START, + MOMENTUM_END, + MOMENTUM_HEADING, + MOMENTUM_MAX_CHARS, + MOMENTUM_START, + append_to_appendix, + appendix_region_bounds, + apply_patch_with_report, + extract_appendix_notes, + momentum_inner, + momentum_region_bounds, + replace_appendix_notes, + replace_momentum, +) + +# ── TestRegionBounds ────────────────────────────────────────────── + + +class TestRegionBounds: + """appendix_region_bounds / momentum_region_bounds 边界与损坏态检测。""" + + # -- appendix -- + + def test_appendix_both_absent_returns_none(self) -> None: + assert appendix_region_bounds("no markers here") is None + + def test_appendix_normal_pair(self) -> None: + text = f"head\n{APPENDIX_START}\nnotes\n{APPENDIX_END}\ntail" + start, end = appendix_region_bounds(text) # type: ignore[misc] + assert text[start:end].startswith(APPENDIX_START) + assert text[start:end].endswith(APPENDIX_END) + + def test_appendix_only_start_raises(self) -> None: + with pytest.raises(ValueError, match="不配对"): + appendix_region_bounds(f"head\n{APPENDIX_START}\nno end") + + def test_appendix_only_end_raises(self) -> None: + with pytest.raises(ValueError, match="不配对"): + appendix_region_bounds(f"head\n{APPENDIX_END}\nno start") + + def test_appendix_repeated_start_raises(self) -> None: + text = f"{APPENDIX_START}\n{APPENDIX_START}\n{APPENDIX_END}" + with pytest.raises(ValueError, match="不配对"): + appendix_region_bounds(text) + + def test_appendix_reversed_raises(self) -> None: + text = f"{APPENDIX_END}\nbody\n{APPENDIX_START}" + with pytest.raises(ValueError, match="不配对"): + appendix_region_bounds(text) + + # -- momentum -- + + def test_momentum_both_absent_returns_none(self) -> None: + assert momentum_region_bounds("no markers here") is None + + def test_momentum_normal_pair(self) -> None: + text = f"head\n{MOMENTUM_START}\nguidance\n{MOMENTUM_END}\ntail" + start, end = momentum_region_bounds(text) # type: ignore[misc] + assert text[start:end].startswith(MOMENTUM_START) + assert text[start:end].endswith(MOMENTUM_END) + + def test_momentum_only_start_raises(self) -> None: + with pytest.raises(ValueError, match="不配对"): + momentum_region_bounds(f"head\n{MOMENTUM_START}\nno end") + + def test_momentum_only_end_raises(self) -> None: + with pytest.raises(ValueError, match="不配对"): + momentum_region_bounds(f"head\n{MOMENTUM_END}\nno start") + + def test_momentum_repeated_end_raises(self) -> None: + text = f"{MOMENTUM_START}\n{MOMENTUM_END}\n{MOMENTUM_END}" + with pytest.raises(ValueError, match="不配对"): + momentum_region_bounds(text) + + def test_momentum_reversed_raises(self) -> None: + text = f"{MOMENTUM_END}\nbody\n{MOMENTUM_START}" + with pytest.raises(ValueError, match="不配对"): + momentum_region_bounds(text) + + +# ── TestAppendix ────────────────────────────────────────────────── + + +class TestAppendix: + """append_to_appendix / extract_appendix_notes / replace_appendix_notes。""" + + def test_append_creates_region(self) -> None: + out = append_to_appendix("# Skill\nbody", ["reminder A"]) + assert APPENDIX_START in out + assert APPENDIX_END in out + assert "- reminder A" in out + + def test_append_accumulates(self) -> None: + step1 = append_to_appendix("# Skill\nbody", ["note 1"]) + step2 = append_to_appendix(step1, ["note 2"]) + assert "- note 1" in step2 + assert "- note 2" in step2 + + def test_append_empty_notes_returns_unchanged(self) -> None: + original = "# Skill\nbody" + assert append_to_appendix(original, []) is original + + def test_append_all_whitespace_notes_returns_unchanged(self) -> None: + original = "# Skill\nbody" + assert append_to_appendix(original, [" ", "\t", ""]) == original + + def test_extract_notes_roundtrip(self) -> None: + content = append_to_appendix("# Skill\nbody", ["aaa", "bbb"]) + notes = extract_appendix_notes(content) + assert notes == ["aaa", "bbb"] + + def test_extract_notes_no_region(self) -> None: + assert extract_appendix_notes("no region") == [] + + def test_replace_notes_overwrites(self) -> None: + content = append_to_appendix("# Skill\nbody", ["old"]) + replaced = replace_appendix_notes(content, ["new1", "new2"]) + notes = extract_appendix_notes(replaced) + assert notes == ["new1", "new2"] + assert "old" not in replaced + + def test_replace_notes_empty_deletes_region(self) -> None: + content = append_to_appendix("# Skill\nbody", ["old"]) + replaced = replace_appendix_notes(content, []) + assert APPENDIX_START not in replaced + assert APPENDIX_END not in replaced + + def test_replace_notes_no_region_creates(self) -> None: + replaced = replace_appendix_notes("# Skill\nbody", ["fresh"]) + assert "- fresh" in replaced + assert APPENDIX_START in replaced + + def test_replace_notes_no_region_empty_notes_unchanged(self) -> None: + original = "# Skill\nbody" + assert replace_appendix_notes(original, []) == original + + def test_append_warns_on_exceeding_max_chars(self, caplog: pytest.LogCaptureFixture) -> None: + """appendix 区超长时 loguru warning,不截断。""" + long_note = "x" * (APPENDIX_MAX_CHARS + 100) + with caplog.at_level("WARNING"): + out = append_to_appendix("body", [long_note]) + assert long_note in out # 不截断 + + +# ── TestMomentum ────────────────────────────────────────────────── + + +class TestMomentum: + """replace_momentum / momentum_inner。""" + + def test_replace_creates_region(self) -> None: + out = replace_momentum("# Skill\nbody", "focus on X") + assert MOMENTUM_START in out + assert MOMENTUM_END in out + assert "focus on X" in out + + def test_replace_overwrites(self) -> None: + step1 = replace_momentum("# Skill\nbody", "old guidance") + step2 = replace_momentum(step1, "new guidance") + assert "new guidance" in step2 + assert "old guidance" not in step2 + + def test_replace_empty_guidance_clears(self) -> None: + """空 guidance 合法:清空旧动量、保留标题和 marker。""" + step1 = replace_momentum("# Skill\nbody", "old guidance") + step2 = replace_momentum(step1, "") + assert MOMENTUM_START in step2 + assert MOMENTUM_END in step2 + assert "old guidance" not in step2 + assert MOMENTUM_HEADING in step2 + + def test_replace_rejects_marker_in_guidance(self) -> None: + with pytest.raises(ValueError, match="marker"): + replace_momentum("body", f"bad {MOMENTUM_START} injection") + + def test_replace_rejects_end_marker_in_guidance(self) -> None: + with pytest.raises(ValueError, match="marker"): + replace_momentum("body", f"bad {MOMENTUM_END} injection") + + def test_momentum_inner_returns_guidance(self) -> None: + content = replace_momentum("# Skill\nbody", "focus on X") + inner = momentum_inner(content) + assert inner == "focus on X" + + def test_momentum_inner_no_region(self) -> None: + assert momentum_inner("no region") == "" + + def test_momentum_inner_strips_heading(self) -> None: + content = replace_momentum("body", "some guidance") + inner = momentum_inner(content) + assert MOMENTUM_HEADING not in inner + assert inner == "some guidance" + + def test_replace_warns_on_exceeding_max_chars(self, caplog: pytest.LogCaptureFixture) -> None: + """momentum 区超长时 loguru warning,不截断。""" + long_guidance = "y" * (MOMENTUM_MAX_CHARS + 100) + with caplog.at_level("WARNING"): + out = replace_momentum("body", long_guidance) + assert long_guidance in out # 不截断 + + def test_coexists_with_appendix(self) -> None: + """momentum 与 appendix 独立共存。""" + content = append_to_appendix("# Skill\nbody", ["note A"]) + content = replace_momentum(content, "focus on X") + assert APPENDIX_START in content + assert APPENDIX_END in content + assert MOMENTUM_START in content + assert MOMENTUM_END in content + notes = extract_appendix_notes(content) + assert notes == ["note A"] + inner = momentum_inner(content) + assert inner == "focus on X" + + +# ── TestApplyPatch ──────────────────────────────────────────────── + + +class TestApplyPatch: + """apply_patch_with_report 的 4 种 op + 冻结区 + 异常处理。""" + + def test_append(self) -> None: + content = "# Title\n\nbody text" + edits = [{"op": "append", "target": "", "content": "new section"}] + out, reports = apply_patch_with_report(content, edits) + assert "new section" in out + assert reports[0]["status"] == "applied_append" + assert reports[0]["index"] == 1 + + def test_insert_after_success(self) -> None: + content = "# Title\n\nanchor line\n\nrest" + edits = [{"op": "insert_after", "target": "anchor line", "content": "inserted"}] + out, reports = apply_patch_with_report(content, edits) + assert "inserted" in out + assert reports[0]["status"] == "applied_insert_after" + + def test_insert_after_missing_target_fallback(self) -> None: + content = "# Title\n\nbody" + edits = [{"op": "insert_after", "target": "nonexistent", "content": "payload"}] + out, reports = apply_patch_with_report(content, edits) + assert "payload" in out + assert reports[0]["status"] == "applied_insert_after_fallback" + + def test_insert_after_protected_skip(self) -> None: + content = "# Title\n\nFROZEN BLOCK\n\nrest" + edits = [{"op": "insert_after", "target": "FROZEN BLOCK", "content": "nope"}] + out, reports = apply_patch_with_report( + content, edits, protected_spans=["FROZEN BLOCK"] + ) + assert out == content + assert reports[0]["status"] == "skipped_protected" + + def test_replace(self) -> None: + content = "# Title\n\nold text\n\nrest" + edits = [{"op": "replace", "target": "old text", "content": "new text"}] + out, reports = apply_patch_with_report(content, edits) + assert "new text" in out + assert "old text" not in out + assert reports[0]["status"] == "applied_replace" + + def test_replace_protected_skip(self) -> None: + content = "# Title\n\nprotected\n\nrest" + edits = [{"op": "replace", "target": "protected", "content": "nope"}] + out, reports = apply_patch_with_report( + content, edits, protected_spans=["protected"] + ) + assert "protected" in out + assert reports[0]["status"] == "skipped_protected" + + def test_delete(self) -> None: + content = "# Title\n\nremove me\n\nrest" + edits = [{"op": "delete", "target": "remove me", "content": ""}] + out, reports = apply_patch_with_report(content, edits) + assert "remove me" not in out + assert reports[0]["status"] == "applied_delete" + + def test_unknown_op(self) -> None: + edits = [{"op": "magic", "target": "x", "content": "y"}] + out, reports = apply_patch_with_report("body", edits) + assert reports[0]["status"] == "skipped_unknown_op" + + def test_non_dict_edit(self) -> None: + edits = ["not a dict"] # type: ignore[list-item] + out, reports = apply_patch_with_report("body", edits) + assert reports[0]["status"] == "error" + assert "非 dict" in reports[0].get("error", "") + + def test_missing_target_for_replace(self) -> None: + edits = [{"op": "replace", "target": "", "content": "payload"}] + out, reports = apply_patch_with_report("body", edits) + assert reports[0]["status"] == "skipped_missing_target" + + def test_missing_target_for_delete(self) -> None: + edits = [{"op": "delete", "target": "", "content": ""}] + out, reports = apply_patch_with_report("body", edits) + assert reports[0]["status"] == "skipped_missing_target" + + def test_report_index_is_1_based(self) -> None: + edits = [ + {"op": "append", "target": "", "content": "a"}, + {"op": "append", "target": "", "content": "b"}, + {"op": "append", "target": "", "content": "c"}, + ] + _, reports = apply_patch_with_report("body", edits) + assert [r["index"] for r in reports] == [1, 2, 3] + + def test_report_truncation(self) -> None: + long_target = "x" * 300 + long_content = "y" * 300 + edits = [{"op": "replace", "target": long_target, "content": long_content}] + _, reports = apply_patch_with_report(long_target, edits) + assert len(reports[0]["target"]) == 200 + assert len(reports[0]["content_preview"]) == 200 + + def test_ranges_recalculated_each_edit(self) -> None: + """冻结区坐标在每条 edit 后重新计算。""" + protected = "FREEZE" + content = f"AAA\n{protected}\nBBB" + edits = [ + {"op": "append", "target": "", "content": "prefix text"}, + {"op": "replace", "target": protected, "content": "nope"}, + ] + out, reports = apply_patch_with_report( + content, edits, protected_spans=[protected] + ) + # append 在 FREEZE 之前插入,坐标右移后 replace 仍能检测冻结区 + assert reports[1]["status"] == "skipped_protected" + + def test_append_before_earliest_protected(self) -> None: + """append 插到 start>0 的最早冻结区之前。""" + content = f"---\nfrontmatter\n---\n\nbody\n\n{APPENDIX_START}\nold\n{APPENDIX_END}" + edits = [{"op": "append", "target": "", "content": "INSERTED"}] + out, reports = apply_patch_with_report( + content, edits, protected_spans=[f"{APPENDIX_START}\nold\n{APPENDIX_END}"] + ) + app_pos = out.find(APPENDIX_START) + ins_pos = out.find("INSERTED") + assert ins_pos < app_pos, "append 应在冻结区之前" + + def test_payload_stripped_target_not_stripped(self) -> None: + """target 不 strip,payload 做 strip。""" + content = " spaced target \nrest" + edits = [ + { + "op": "replace", + "target": " spaced target ", + "content": " trimmed ", + } + ] + out, reports = apply_patch_with_report(content, edits) + # payload stripped → "trimmed" + assert "trimmed" in out + assert " trimmed " not in out + assert reports[0]["status"] == "applied_replace" + + def test_replace_count_one(self) -> None: + """replace 只替换第一次出现。""" + content = "dup\ndup\ndup" + edits = [{"op": "replace", "target": "dup", "content": "REPLACED"}] + out, _ = apply_patch_with_report(content, edits) + assert out.count("REPLACED") == 1 + assert out.count("dup") == 2 + + def test_multiple_edits_sequential(self) -> None: + """多条 edit 顺序执行。""" + content = "line1\nline2\nline3" + edits = [ + {"op": "replace", "target": "line1", "content": "LINE_ONE"}, + {"op": "delete", "target": "line2", "content": ""}, + {"op": "append", "target": "", "content": "TAIL"}, + ] + out, reports = apply_patch_with_report(content, edits) + assert "LINE_ONE" in out + assert "line2" not in out + assert "TAIL" in out + assert all(r["status"].startswith("applied") for r in reports)