fix: patch checks full target span + marker injection/integrity (algo #8)
This commit is contained in:
@@ -19,6 +19,8 @@ from loguru import logger
|
||||
from core.evolution.patch import (
|
||||
APPENDIX_END,
|
||||
APPENDIX_START,
|
||||
MOMENTUM_END,
|
||||
MOMENTUM_START,
|
||||
append_to_appendix,
|
||||
apply_patch_with_report,
|
||||
extract_appendix_notes,
|
||||
@@ -293,10 +295,39 @@ def _tool_protected_spans(text: str) -> list[str]:
|
||||
# =========================================================================
|
||||
|
||||
|
||||
def _check_marker_integrity(evolved: str) -> list[str]:
|
||||
"""校验 evolved 中冻结区 marker 的完整性（成对、至多一对、START 先于 END）。
|
||||
|
||||
进化写入可能破坏 appendix/momentum marker 配对，破坏后 append_to_appendix /
|
||||
replace_momentum 等下游会静默误拼或抛错。此处集中拦截：任一 marker 对违反
|
||||
「START 数==END 数、各至多一对、START 在 END 前」即整体 reject。
|
||||
|
||||
参数:
|
||||
evolved: 改写后的全文。
|
||||
返回:
|
||||
错误信息列表（空列表表示 marker 完整）。
|
||||
"""
|
||||
errors: list[str] = []
|
||||
for name, start_m, end_m in (
|
||||
("APPENDIX", APPENDIX_START, APPENDIX_END),
|
||||
("MOMENTUM", MOMENTUM_START, MOMENTUM_END),
|
||||
):
|
||||
s = evolved.count(start_m)
|
||||
e = evolved.count(end_m)
|
||||
if s != e:
|
||||
errors.append(f"{name} marker 不配对：START={s} END={e}")
|
||||
elif s > 1:
|
||||
errors.append(f"{name} marker 出现多对（{s}），至多一对")
|
||||
elif s == 1 and evolved.index(start_m) > evolved.index(end_m):
|
||||
errors.append(f"{name} marker 顺序错误：END 出现在 START 之前")
|
||||
return errors
|
||||
|
||||
|
||||
def validate_skill(original: str, evolved: str) -> ValidationResult:
|
||||
"""校验 Skill 改写结果。
|
||||
|
||||
检查项: frontmatter 三字段保留（name / description / task_type）、
|
||||
marker 完整性（appendix/momentum 成对且至多一对、顺序正确）、
|
||||
长度比在 [0.3, 2.0]、代码块闭合。
|
||||
|
||||
参数:
|
||||
@@ -305,6 +336,11 @@ def validate_skill(original: str, evolved: str) -> ValidationResult:
|
||||
|
||||
返回:
|
||||
ValidationResult 实例。
|
||||
|
||||
关键实现细节:
|
||||
marker 完整性先于长度校验——长度校验经 _strip_protected_regions 调用
|
||||
momentum_region_bounds，对损坏 marker 会抛 ValueError；故 marker 破坏时先
|
||||
返回失败，避免异常穿透且明确 reject 该候选。
|
||||
"""
|
||||
errors: list[str] = []
|
||||
orig_fm = _parse_frontmatter(original)
|
||||
@@ -319,6 +355,10 @@ def validate_skill(original: str, evolved: str) -> ValidationResult:
|
||||
errors.append(
|
||||
f"frontmatter 字段 {key} 被修改: {orig_fm.get(key)!r} → {evol_fm.get(key)!r}"
|
||||
)
|
||||
marker_errors = _check_marker_integrity(evolved)
|
||||
if marker_errors:
|
||||
errors.extend(marker_errors)
|
||||
return ValidationResult(passed=False, errors=errors)
|
||||
errors.extend(_check_length(original, evolved))
|
||||
errors.extend(_check_code_blocks(evolved))
|
||||
return ValidationResult(passed=len(errors) == 0, errors=errors)
|
||||
|
||||
+43
-7
@@ -282,9 +282,34 @@ def _protected_ranges(content: str, spans: list[str]) -> list[tuple[int, int]]:
|
||||
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 _span_overlaps_ranges(pos: int, length: int, ranges: list[tuple[int, int]]) -> bool:
|
||||
"""判断 [pos, pos+length) 是否与任一冻结区间相交（不止起点）。
|
||||
|
||||
起点落在正文、末端伸入冻结区的 target 也须拦截，否则 replace/delete 会连带
|
||||
改动冻结区（如破坏 appendix/momentum marker）。
|
||||
|
||||
参数:
|
||||
pos: target 在正文中的起点。
|
||||
length: target 长度。
|
||||
ranges: 冻结区间 [start, end) 列表。
|
||||
返回:
|
||||
与任一区间相交返回 True。
|
||||
"""
|
||||
end = pos + length
|
||||
return any(start < end and pos < r_end for start, r_end in ranges)
|
||||
|
||||
|
||||
# 冻结区 marker 字面量：LLM 生成的 edit 不得注入这些字面量，否则破坏 marker 配对
|
||||
_MARKER_LITERALS = (APPENDIX_START, APPENDIX_END, MOMENTUM_START, MOMENTUM_END)
|
||||
|
||||
|
||||
def _edit_injects_marker(edit: dict) -> bool:
|
||||
"""判断 edit 的 target/content 是否含冻结区 marker 字面量（注入拦截）。"""
|
||||
for key in ("target", "content"):
|
||||
value = edit.get(key)
|
||||
if isinstance(value, str) and any(m in value for m in _MARKER_LITERALS):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _append_at(content: str, ranges: list[tuple[int, int]]) -> int:
|
||||
@@ -317,8 +342,8 @@ def _do_insert_after(
|
||||
_insert_at(content, _append_at(content, ranges), payload),
|
||||
"applied_insert_after_fallback",
|
||||
)
|
||||
if _in_ranges(pos, ranges):
|
||||
logger.warning("insert_after 目标在冻结区，跳过 target={}", target[:80])
|
||||
if _span_overlaps_ranges(pos, len(target), ranges):
|
||||
logger.warning("insert_after 目标跨入冻结区，跳过 target={}", target[:80])
|
||||
return content, "skipped_protected"
|
||||
at = pos + len(target)
|
||||
nl = content.find("\n", at)
|
||||
@@ -340,8 +365,8 @@ def _do_replace_delete(
|
||||
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])
|
||||
if _span_overlaps_ranges(pos, len(target), 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
|
||||
@@ -403,6 +428,17 @@ def apply_patch_with_report(
|
||||
reports: list[dict] = []
|
||||
for i, edit in enumerate(edits, 1):
|
||||
try:
|
||||
if isinstance(edit, dict) and _edit_injects_marker(edit):
|
||||
logger.warning("edit 含冻结区 marker 字面量，拒绝该 edit index={}", i)
|
||||
report = {
|
||||
"op": str(edit.get("op", "")),
|
||||
"target": str(edit.get("target", "") or "")[:200],
|
||||
"content_preview": str(edit.get("content", "") or "")[:200],
|
||||
"status": "skipped_marker_injection",
|
||||
}
|
||||
report["index"] = i
|
||||
reports.append(report)
|
||||
continue
|
||||
ranges = _protected_ranges(content, spans)
|
||||
content, report = _apply_one(content, edit, ranges)
|
||||
except (KeyError, TypeError, ValueError, AttributeError) as exc:
|
||||
|
||||
@@ -309,6 +309,36 @@ class TestValidateSkill:
|
||||
result = validate_skill(orig, "no frontmatter body")
|
||||
assert not result.passed
|
||||
|
||||
def test_unpaired_appendix_marker_fails(self) -> None:
|
||||
"""evolved 出现孤立 APPENDIX_START（无配对 END）→ marker 完整性校验失败。"""
|
||||
orig = "---\nname: a\ndescription: d\ntask_type: t\n---\nbody body body"
|
||||
evol = f"---\nname: a\ndescription: d\ntask_type: t\n---\nbody {APPENDIX_START} body"
|
||||
result = validate_skill(orig, evol)
|
||||
assert not result.passed
|
||||
assert any("marker" in e.lower() or "配对" in e for e in result.errors)
|
||||
|
||||
def test_duplicate_momentum_marker_fails(self) -> None:
|
||||
"""evolved 出现两对 MOMENTUM marker → 完整性校验失败（各至多一对）。"""
|
||||
orig = "---\nname: a\ndescription: d\ntask_type: t\n---\nbody body body"
|
||||
evol = (
|
||||
"---\nname: a\ndescription: d\ntask_type: t\n---\n"
|
||||
f"{MOMENTUM_START}x{MOMENTUM_END} mid {MOMENTUM_START}y{MOMENTUM_END}"
|
||||
)
|
||||
result = validate_skill(orig, evol)
|
||||
assert not result.passed
|
||||
assert any("marker" in e.lower() or "配对" in e for e in result.errors)
|
||||
|
||||
def test_paired_markers_pass(self) -> None:
|
||||
"""evolved 含成对 appendix+momentum marker（各一对，顺序正确）→ 校验通过。"""
|
||||
body = "body " * 20
|
||||
orig = f"---\nname: a\ndescription: d\ntask_type: t\n---\n{body}"
|
||||
evol = (
|
||||
f"---\nname: a\ndescription: d\ntask_type: t\n---\n{body}"
|
||||
f"{APPENDIX_START}\n- n\n{APPENDIX_END}\n{MOMENTUM_START}\nm\n{MOMENTUM_END}"
|
||||
)
|
||||
result = validate_skill(orig, evol)
|
||||
assert result.passed, result.errors
|
||||
|
||||
|
||||
class TestValidateSystem:
|
||||
"""validate_system 测试。"""
|
||||
|
||||
@@ -383,3 +383,68 @@ class TestApplyPatch:
|
||||
assert "line2" not in out
|
||||
assert "TAIL" in out
|
||||
assert all(r["status"].startswith("applied") for r in reports)
|
||||
|
||||
|
||||
class TestSpanOverlapProtection:
|
||||
"""冻结区跨度保护 + marker 注入拦截（算法 #8 加固）。"""
|
||||
|
||||
def test_replace_spanning_into_protected_marker_preserved(self) -> None:
|
||||
"""target 末端跨入含 marker 的 appendix 冻结区时被拦截，marker 不被破坏。"""
|
||||
body = "正文最后一段。"
|
||||
appendix = f"{APPENDIX_START}\n## 执行提醒\n- 规则A\n{APPENDIX_END}"
|
||||
content = body + "\n\n" + appendix
|
||||
# target 从正文末尾跨入 APPENDIX_START（含 marker 字面量）
|
||||
target = "正文最后一段。\n\n" + APPENDIX_START
|
||||
edits = [{"op": "delete", "target": target, "content": ""}]
|
||||
new_content, report = apply_patch_with_report(
|
||||
content, edits, protected_spans=[appendix]
|
||||
)
|
||||
# 无论经 marker 注入拦截还是跨度拦截，marker 都必须完整保留
|
||||
assert APPENDIX_START in new_content and APPENDIX_END in new_content
|
||||
assert report[0]["status"] in ("skipped_protected", "skipped_marker_injection")
|
||||
|
||||
def test_replace_spanning_into_frozen_section_is_skipped(self) -> None:
|
||||
"""target 起点在正文、末端伸入无 marker 的冻结区段 → 跨度拦截跳过。"""
|
||||
body = "可改正文段落。"
|
||||
frozen = "## 输出格式\n必须输出 JSON。"
|
||||
content = body + "\n" + frozen
|
||||
# target 从可改正文跨入冻结区段（不含任何 marker 字面量）
|
||||
target = "可改正文段落。\n## 输出格式"
|
||||
edits = [{"op": "replace", "target": target, "content": "破坏内容"}]
|
||||
new_content, report = apply_patch_with_report(
|
||||
content, edits, protected_spans=[frozen]
|
||||
)
|
||||
assert "破坏内容" not in new_content
|
||||
assert "## 输出格式" in new_content
|
||||
assert report[0]["status"] == "skipped_protected"
|
||||
|
||||
def test_insert_after_spanning_into_frozen_section_is_skipped(self) -> None:
|
||||
"""insert_after 的 target 末端伸入无 marker 冻结区段时跨度拦截跳过。"""
|
||||
body = "可改正文。"
|
||||
frozen = "## 输出格式\n固定内容。"
|
||||
content = body + "\n" + frozen
|
||||
target = "可改正文。\n## 输出格式"
|
||||
edits = [{"op": "insert_after", "target": target, "content": "注入内容"}]
|
||||
new_content, report = apply_patch_with_report(
|
||||
content, edits, protected_spans=[frozen]
|
||||
)
|
||||
assert "注入内容" not in new_content
|
||||
assert report[0]["status"] == "skipped_protected"
|
||||
|
||||
def test_edit_payload_with_marker_literal_rejected(self) -> None:
|
||||
"""edit payload 含 marker 字面量 → 拒绝该 edit。"""
|
||||
edits = [
|
||||
{"op": "append", "target": "", "content": f"注入 {APPENDIX_START} 破坏"}
|
||||
]
|
||||
_, report = apply_patch_with_report("正文", edits, protected_spans=[])
|
||||
assert any(
|
||||
"marker" in str(s).lower() or "reject" in str(s).lower() for s in report[0].values()
|
||||
)
|
||||
|
||||
def test_edit_target_with_marker_literal_rejected(self) -> None:
|
||||
"""edit target 含 momentum marker 字面量 → 拒绝该 edit。"""
|
||||
edits = [
|
||||
{"op": "replace", "target": f"{MOMENTUM_START}x", "content": "y"}
|
||||
]
|
||||
_, report = apply_patch_with_report("正文", edits, protected_spans=[])
|
||||
assert report[0]["status"] == "skipped_marker_injection"
|
||||
|
||||
Reference in New Issue
Block a user