iomgaa
3005f24577
从 TRM4 core/harness/patch.py (427 行) 零修改移植到 core/evolution/patch.py。 包含: - 7 个常量（APPENDIX/MOMENTUM marker + MAX_CHARS + HEADING） - appendix 区：追加/提取/替换/边界检测，损坏态 ValueError - momentum 区：替换/提取/边界检测，注入防护 - apply_patch_with_report：4 种 op（append/insert_after/replace/delete） - 冻结区坐标判定（每条 edit 重算）、report 1-based index - 51 个单元测试覆盖全部公共 API 及边缘场景 Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
428 lines
18 KiB
Python
428 lines
18 KiB
Python
"""定点补丁引擎：把进化输出的离散 edit 逐条应用到文本，逐条出状态报告。
|
||
|
||
借鉴 SkillOpt skill.py 的 apply 语义；守 P5：找不到锚点不静默乱改、不裸 except。
|
||
冻结区按全文坐标区间判定；append/退化追加插到最早冻结区之前（无则 EOF）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from loguru import logger
|
||
|
||
APPENDIX_START = "<!-- APPENDIX_START -->"
|
||
APPENDIX_END = "<!-- APPENDIX_END -->"
|
||
APPENDIX_MAX_CHARS = 2000 # appendix 区软上限（守设计「长度上限+warning，不做去重」）
|
||
|
||
MOMENTUM_START = "<!-- MOMENTUM_START -->"
|
||
MOMENTUM_END = "<!-- 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
|