601 lines
22 KiB
Python
601 lines
22 KiB
Python
"""Validate and apply source-grounded semantic rewrite plans."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections import Counter, defaultdict
|
|
import re
|
|
from typing import Any
|
|
|
|
from .document import (
|
|
CANONICAL_HEADINGS,
|
|
HEADING_RE,
|
|
INLINE_PROTECTED_RE,
|
|
LIST_RE,
|
|
SkillDocument,
|
|
parse_document,
|
|
section_key,
|
|
)
|
|
from .models import Operation, SemanticRewriteUnit, SourceRef
|
|
|
|
|
|
PLAN_SCHEMA_VERSION = "2.0"
|
|
ALLOWED_KINDS = {"add_summary", "replace_block"}
|
|
# Tables and code samples may contain explicit operational constraints. They
|
|
# are read-only evidence: usable for summaries, never as replacement targets.
|
|
SUMMARY_SOURCE_KINDS = frozenset({"paragraph", "list_item", "table", "code"})
|
|
REPLACE_SOURCE_KINDS = frozenset({"paragraph", "list_item"})
|
|
ALLOWED_SECTIONS = {
|
|
"definitions",
|
|
"critical_rules",
|
|
"evidence_priority",
|
|
"scope",
|
|
"decision_criteria",
|
|
"uncertainty_rule",
|
|
"task",
|
|
"inputs",
|
|
"output",
|
|
"validation",
|
|
"completion_criterion",
|
|
}
|
|
SECTION_ORDER = (
|
|
"definitions",
|
|
"critical_rules",
|
|
"evidence_priority",
|
|
"scope",
|
|
"decision_criteria",
|
|
"uncertainty_rule",
|
|
"inputs",
|
|
"task",
|
|
"output",
|
|
"validation",
|
|
"completion_criterion",
|
|
)
|
|
HARD_PROTECTED_RE = re.compile(
|
|
INLINE_PROTECTED_RE.pattern
|
|
+ r"|\b\d+(?:\.\d+)*%?\b"
|
|
+ r"|\b(?:MUST(?:\s+NOT)?|SHALL(?:\s+NOT)?|SHOULD(?:\s+NOT)?|"
|
|
+ r"NEVER|ALWAYS|DO\s+NOT|ONLY|"
|
|
+ r"IF|WHEN|UNLESS|BEFORE|AFTER|EXCEPT|WITHOUT)\b"
|
|
+ r"|不得|必须|禁止|仅可|只能|不能|不要|始终|如果|当|除非|之前|之后|除外|不得不",
|
|
re.I,
|
|
)
|
|
SOFT_MODAL_RE = re.compile(r"\b(?:MAY(?:\s+NOT)?|CAN(?:\s+NOT)?)\b", re.I)
|
|
PROTECTED_RE = re.compile(
|
|
HARD_PROTECTED_RE.pattern + r"|" + SOFT_MODAL_RE.pattern,
|
|
re.I,
|
|
)
|
|
LIST_PREFIX_RE = re.compile(r"^([ \t]*(?:[-+*]|\d+[.)])[ \t]+)")
|
|
ORDERED_LIST_PREFIX_RE = re.compile(r"^[ \t]*\d+[.)][ \t]+", re.M)
|
|
EXPLICIT_RULE_RE = re.compile(
|
|
r"\b(?:MUST(?:\s+NOT)?|SHALL(?:\s+NOT)?|SHOULD(?:\s+NOT)?|"
|
|
r"MAY(?:\s+NOT)?|NEVER|ALWAYS|DO\s+NOT|ONLY|REQUIRED|PROHIBITED|"
|
|
r"IF|WHEN|UNLESS|BEFORE|AFTER|EXCEPT|WITHOUT)\b"
|
|
r"|\b(?:is|are|was|were|do|does|can|will|they(?:'re|\s+are))\s+not\b"
|
|
r"|(?:^|\n|\s-\s)(?:keep|remove|use|read|write|recurse|work|check|"
|
|
r"validate|ensure|avoid|preserve|process|return|run|call|set|include|"
|
|
r"exclude)\b(?!\s+\d+\s*:)"
|
|
r"|不得|必须|禁止|仅可|只能|不能|不要|始终|如果|当|除非|之前|之后|"
|
|
r"应当|需要|务必|确保|保留|删除|移除|使用|读取|写入|检查|验证",
|
|
re.I,
|
|
)
|
|
RESOURCE_CONFIG_ASSIGNMENT_RE = re.compile(
|
|
r"^\s*(?:export\s+)?[A-Za-z_][A-Za-z0-9_]*(?:PATH|DIR|CACHE|ROOT|HOME|"
|
|
r"CONFIG|ENDPOINT|HOST|PORT|MODE|OFFLINE|DATABASE|DB)[A-Za-z0-9_]*\s*=\s*"
|
|
r"(?:[rRuUbBfF]{0,2})?['\"][^'\"]+['\"]\s*$",
|
|
re.I,
|
|
)
|
|
RESOURCE_PARAMETER_RE = re.compile(
|
|
r"`--[A-Za-z0-9][A-Za-z0-9-]*\s+<[^>]+>`.*"
|
|
r"\b(?:path|directory|dir|cache|database|db|location|file)\b",
|
|
re.I,
|
|
)
|
|
NARROW_TASK_HEADING_RE = re.compile(
|
|
r"\b(?:conditional|condition|branch|pattern|example|implementation|detail|"
|
|
r"substep|edge case)s?\b|条件|分支|模式|示例|实现细节|子步骤|边界情况",
|
|
re.I,
|
|
)
|
|
REJECTION_INDEX_RE = re.compile(r"^rewrite\[(\d+)]\s*:")
|
|
REPAIRABLE_REJECTION_MARKERS = (
|
|
"replacement is not text",
|
|
"replacement is empty",
|
|
"replacement may not inject headings or fenced code",
|
|
"replacement is disproportionately longer than its sources",
|
|
"changed a protected literal, number, or modality",
|
|
"block rewrite is too short",
|
|
"block rewrite changed paragraph/list structure",
|
|
"block rewrite changed the list marker",
|
|
"summary changed a hard protected source literal",
|
|
)
|
|
|
|
|
|
def semantic_plan_needed(document: SkillDocument) -> bool:
|
|
return any(
|
|
block.kind in SUMMARY_SOURCE_KINDS and block.text.strip()
|
|
for block in document.blocks
|
|
)
|
|
|
|
|
|
def rejection_index(reason: str) -> int | None:
|
|
match = REJECTION_INDEX_RE.match(reason)
|
|
return int(match.group(1)) if match else None
|
|
|
|
|
|
def repairable_rewrite_indices(reasons: list[str]) -> list[int]:
|
|
indices = []
|
|
for reason in reasons:
|
|
index = rejection_index(reason)
|
|
if index is not None and any(
|
|
marker in reason for marker in REPAIRABLE_REJECTION_MARKERS
|
|
):
|
|
indices.append(index)
|
|
return indices
|
|
|
|
|
|
def _protected(
|
|
text: str,
|
|
*,
|
|
ignore_list_ordinals: bool = False,
|
|
hard_only: bool = False,
|
|
) -> Counter[str]:
|
|
if ignore_list_ordinals:
|
|
# Ordered-list markers describe Markdown structure, not task semantics.
|
|
# Remove only line-leading markers; numbers in the item body remain
|
|
# protected (for example, "Retry 3 times" or "use version 2.1").
|
|
text = ORDERED_LIST_PREFIX_RE.sub("", text)
|
|
pattern = HARD_PROTECTED_RE if hard_only else PROTECTED_RE
|
|
return Counter(match.group(0).casefold() for match in pattern.finditer(text))
|
|
|
|
|
|
def _soft_modal_change(source: str, replacement: str) -> str | None:
|
|
source_values = Counter(
|
|
match.group(0).casefold() for match in SOFT_MODAL_RE.finditer(source)
|
|
)
|
|
replacement_values = Counter(
|
|
match.group(0).casefold() for match in SOFT_MODAL_RE.finditer(replacement)
|
|
)
|
|
if set(source_values) == set(replacement_values):
|
|
return None
|
|
return _protected_change(source_values, replacement_values, compare_counts=False)
|
|
|
|
|
|
def _protected_change(
|
|
source: Counter[str], replacement: Counter[str], *, compare_counts: bool
|
|
) -> str:
|
|
if compare_counts:
|
|
missing = sorted((source - replacement).elements())
|
|
added = sorted((replacement - source).elements())
|
|
else:
|
|
missing = sorted(set(source) - set(replacement))
|
|
added = sorted(set(replacement) - set(source))
|
|
details = []
|
|
if missing:
|
|
details.append(f"missing={missing!r}")
|
|
if added:
|
|
details.append(f"added={added!r}")
|
|
return ", ".join(details) or "no difference"
|
|
|
|
|
|
def _source_material(refs: tuple[SourceRef, ...]) -> str:
|
|
return "\n".join(ref.quote for ref in refs)
|
|
|
|
|
|
def _literal_map(
|
|
text: str,
|
|
*,
|
|
ignore_list_ordinals: bool = False,
|
|
hard_only: bool = False,
|
|
soft_only: bool = False,
|
|
) -> dict[str, str]:
|
|
if ignore_list_ordinals:
|
|
text = ORDERED_LIST_PREFIX_RE.sub("", text)
|
|
if soft_only:
|
|
pattern = SOFT_MODAL_RE
|
|
elif hard_only:
|
|
pattern = HARD_PROTECTED_RE
|
|
else:
|
|
pattern = PROTECTED_RE
|
|
values: dict[str, str] = {}
|
|
for match in pattern.finditer(text):
|
|
values.setdefault(match.group(0).casefold(), match.group(0))
|
|
return values
|
|
|
|
|
|
def replacement_literal_delta(
|
|
raw: dict[str, Any], document: SkillDocument
|
|
) -> dict[str, list[str]]:
|
|
"""Return structured literal edits for a locked replacement repair."""
|
|
|
|
raw_refs = raw.get("source_refs")
|
|
replacement = raw.get("replacement")
|
|
kind = raw.get("kind")
|
|
if not isinstance(raw_refs, list) or not isinstance(replacement, str):
|
|
return {
|
|
"restore_verbatim": [],
|
|
"remove_verbatim": [],
|
|
"soft_modal_missing": [],
|
|
"soft_modal_added": [],
|
|
}
|
|
quotes = [
|
|
ref.get("quote")
|
|
for ref in raw_refs
|
|
if isinstance(ref, dict) and isinstance(ref.get("quote"), str)
|
|
]
|
|
if len(quotes) != len(raw_refs):
|
|
return {
|
|
"restore_verbatim": [],
|
|
"remove_verbatim": [],
|
|
"soft_modal_missing": [],
|
|
"soft_modal_added": [],
|
|
}
|
|
source = "\n".join(quotes)
|
|
summary = kind == "add_summary"
|
|
source_hard = _literal_map(
|
|
source,
|
|
ignore_list_ordinals=summary,
|
|
hard_only=summary,
|
|
)
|
|
replacement_hard = _literal_map(
|
|
replacement,
|
|
ignore_list_ordinals=summary,
|
|
hard_only=summary,
|
|
)
|
|
source_soft = _literal_map(source, soft_only=True)
|
|
replacement_soft = _literal_map(replacement, soft_only=True)
|
|
return {
|
|
"restore_verbatim": [
|
|
source_hard[key] for key in sorted(set(source_hard) - set(replacement_hard))
|
|
],
|
|
"remove_verbatim": [
|
|
replacement_hard[key]
|
|
for key in sorted(set(replacement_hard) - set(source_hard))
|
|
],
|
|
"soft_modal_missing": [
|
|
source_soft[key] for key in sorted(set(source_soft) - set(replacement_soft))
|
|
],
|
|
"soft_modal_added": [
|
|
replacement_soft[key]
|
|
for key in sorted(set(replacement_soft) - set(source_soft))
|
|
],
|
|
}
|
|
|
|
|
|
def _summary_target_error(
|
|
target: str, source: str, refs: list[SourceRef], blocks: dict[str, Any]
|
|
) -> str | None:
|
|
resource_parameter_binding = (
|
|
any(
|
|
blocks[ref.block_id].kind == "code"
|
|
and RESOURCE_CONFIG_ASSIGNMENT_RE.match(ref.quote)
|
|
for ref in refs
|
|
)
|
|
and any(RESOURCE_PARAMETER_RE.search(ref.quote) for ref in refs)
|
|
)
|
|
if (
|
|
target == "critical_rules"
|
|
and not EXPLICIT_RULE_RE.search(source)
|
|
and not resource_parameter_binding
|
|
):
|
|
return (
|
|
"critical_rules summary source is descriptive rather than an explicit "
|
|
"directive, prohibition, condition, or required invariant"
|
|
)
|
|
if target == "task":
|
|
headings = [blocks[ref.block_id].parent_heading for ref in refs]
|
|
if headings and all(
|
|
heading and NARROW_TASK_HEADING_RE.search(heading)
|
|
for heading in headings
|
|
):
|
|
return "task summary may not promote a narrow subsection into the global task"
|
|
return None
|
|
|
|
|
|
def _validate_replacement_shape(
|
|
kind: str, source: str, replacement: str
|
|
) -> str | None:
|
|
if not replacement.strip():
|
|
return "replacement is empty"
|
|
if "```" in replacement or "~~~" in replacement or re.search(
|
|
r"^#{1,6}[ \t]+", replacement, re.M
|
|
):
|
|
return "replacement may not inject headings or fenced code"
|
|
if len(replacement) > max(400, int(len(source) * 1.75)):
|
|
return "replacement is disproportionately longer than its sources"
|
|
if kind == "replace_block":
|
|
source_protected = _protected(source)
|
|
replacement_protected = _protected(replacement)
|
|
if source_protected != replacement_protected:
|
|
change = _protected_change(
|
|
source_protected, replacement_protected, compare_counts=True
|
|
)
|
|
return (
|
|
"block rewrite changed a protected literal, number, or modality "
|
|
f"({change})"
|
|
)
|
|
ratio = len(replacement.strip()) / max(1, len(source.strip()))
|
|
if ratio < 0.55:
|
|
return "block rewrite is too short to preserve all source content"
|
|
source_prefix = LIST_PREFIX_RE.match(source)
|
|
replacement_prefix = LIST_PREFIX_RE.match(replacement)
|
|
if bool(source_prefix) != bool(replacement_prefix):
|
|
return "block rewrite changed paragraph/list structure"
|
|
if source_prefix and replacement_prefix:
|
|
if source_prefix.group(1) != replacement_prefix.group(1):
|
|
return "block rewrite changed the list marker"
|
|
else:
|
|
source_protected = _protected(
|
|
source, ignore_list_ordinals=True, hard_only=True
|
|
)
|
|
replacement_protected = _protected(
|
|
replacement, ignore_list_ordinals=True, hard_only=True
|
|
)
|
|
if set(source_protected) != set(replacement_protected):
|
|
change = _protected_change(
|
|
source_protected, replacement_protected, compare_counts=False
|
|
)
|
|
soft_change = _soft_modal_change(source, replacement)
|
|
if soft_change:
|
|
change += f"; soft_modal_change=({soft_change})"
|
|
return (
|
|
"summary changed a hard protected source literal "
|
|
f"({change})"
|
|
)
|
|
return None
|
|
|
|
|
|
def validate_semantic_plan(
|
|
payload: dict[str, Any], document: SkillDocument
|
|
) -> tuple[list[SemanticRewriteUnit], list[str]]:
|
|
if payload.get("schema_version") != PLAN_SCHEMA_VERSION:
|
|
raise ValueError("semantic plan has unsupported schema_version")
|
|
raw_units = payload.get("rewrites")
|
|
if not isinstance(raw_units, list):
|
|
raise ValueError("semantic plan requires rewrites list")
|
|
if len(raw_units) > 24:
|
|
raise ValueError("semantic plan exceeds 24 rewrite units")
|
|
blocks = document.block_index
|
|
accepted: list[SemanticRewriteUnit] = []
|
|
rejected: list[str] = []
|
|
replaced_blocks: set[str] = set()
|
|
for index, raw in enumerate(raw_units):
|
|
prefix = f"rewrite[{index}]"
|
|
if not isinstance(raw, dict):
|
|
rejected.append(f"{prefix}: item is not an object")
|
|
continue
|
|
kind = raw.get("kind")
|
|
target = raw.get("target_section")
|
|
replacement = raw.get("replacement")
|
|
confidence = raw.get("confidence")
|
|
raw_refs = raw.get("source_refs")
|
|
if kind not in ALLOWED_KINDS:
|
|
rejected.append(f"{prefix}: unsupported kind")
|
|
continue
|
|
if kind == "add_summary" and target not in ALLOWED_SECTIONS:
|
|
rejected.append(f"{prefix}: unsupported target_section")
|
|
continue
|
|
if kind == "replace_block":
|
|
target = None
|
|
if not isinstance(replacement, str):
|
|
rejected.append(f"{prefix}: replacement is not text")
|
|
continue
|
|
if (
|
|
isinstance(confidence, bool)
|
|
or not isinstance(confidence, (int, float))
|
|
or float(confidence) < (0.90 if kind == "replace_block" else 0.85)
|
|
):
|
|
rejected.append(f"{prefix}: invalid confidence")
|
|
continue
|
|
if not isinstance(raw_refs, list) or not 1 <= len(raw_refs) <= 8:
|
|
rejected.append(f"{prefix}: invalid source_refs")
|
|
continue
|
|
refs: list[SourceRef] = []
|
|
invalid_ref = False
|
|
non_prose_replacement_ref = False
|
|
for raw_ref in raw_refs:
|
|
if not isinstance(raw_ref, dict):
|
|
invalid_ref = True
|
|
break
|
|
block_id = raw_ref.get("block_id")
|
|
quote = raw_ref.get("quote")
|
|
block = blocks.get(block_id) if isinstance(block_id, str) else None
|
|
if (
|
|
block is None
|
|
or block.kind not in SUMMARY_SOURCE_KINDS
|
|
or not isinstance(quote, str)
|
|
or not quote
|
|
or quote not in block.text
|
|
):
|
|
invalid_ref = True
|
|
break
|
|
if kind == "replace_block" and block.kind not in REPLACE_SOURCE_KINDS:
|
|
non_prose_replacement_ref = True
|
|
break
|
|
refs.append(SourceRef(block_id, quote))
|
|
if invalid_ref:
|
|
rejected.append(f"{prefix}: source_refs are not exact prose spans")
|
|
continue
|
|
if non_prose_replacement_ref:
|
|
rejected.append(
|
|
f"{prefix}: replace_block may only cite paragraph or list_item sources"
|
|
)
|
|
continue
|
|
if kind == "replace_block":
|
|
if len(refs) != 1 or refs[0].quote != blocks[refs[0].block_id].text:
|
|
rejected.append(f"{prefix}: replace_block must cite one complete block")
|
|
continue
|
|
if refs[0].block_id in replaced_blocks:
|
|
rejected.append(f"{prefix}: block already has an accepted replacement")
|
|
continue
|
|
source = _source_material(tuple(refs))
|
|
if kind == "add_summary":
|
|
target_error = _summary_target_error(target, source, refs, blocks)
|
|
if target_error:
|
|
rejected.append(f"{prefix}: {target_error}")
|
|
continue
|
|
shape_error = _validate_replacement_shape(kind, source, replacement)
|
|
if shape_error:
|
|
rejected.append(f"{prefix}: {shape_error}")
|
|
continue
|
|
unit = SemanticRewriteUnit(
|
|
kind=kind,
|
|
target_section=target,
|
|
source_refs=tuple(refs),
|
|
replacement=replacement.strip(),
|
|
confidence=float(confidence),
|
|
)
|
|
accepted.append(unit)
|
|
if kind == "replace_block":
|
|
replaced_blocks.add(refs[0].block_id)
|
|
return accepted, rejected
|
|
|
|
|
|
def _heading_for(key: str, language: str) -> str:
|
|
return CANONICAL_HEADINGS[language][key]
|
|
|
|
|
|
def _section_text(document: SkillDocument, heading: Any) -> str:
|
|
"""Return only the content governed by a recognized heading."""
|
|
|
|
end = len(document.body)
|
|
level = heading.heading_level or 6
|
|
for block in document.blocks:
|
|
if (
|
|
block.kind == "heading"
|
|
and block.start_offset > heading.start_offset
|
|
and (block.heading_level or 6) <= level
|
|
):
|
|
end = block.start_offset
|
|
break
|
|
return document.body[heading.end_offset:end]
|
|
|
|
|
|
def apply_semantic_plan(
|
|
content: str,
|
|
source_document: SkillDocument,
|
|
units: list[SemanticRewriteUnit],
|
|
) -> tuple[str, list[Operation], list[str]]:
|
|
"""Apply valid units independently; return skip reasons for local fallback."""
|
|
|
|
body = parse_document(content).body
|
|
frontmatter = parse_document(content).frontmatter
|
|
operations: list[Operation] = []
|
|
skipped: list[str] = []
|
|
patches: list[tuple[int, int, str]] = []
|
|
summary_values: dict[str, list[tuple[SemanticRewriteUnit, str]]] = defaultdict(list)
|
|
|
|
for index, unit in enumerate(units):
|
|
if unit.kind == "add_summary":
|
|
assert unit.target_section is not None
|
|
value = unit.replacement
|
|
if not re.match(r"^[-+*][ \t]+", value):
|
|
value = f"- {value}"
|
|
summary_values[unit.target_section].append((unit, value))
|
|
continue
|
|
ref = unit.source_refs[0]
|
|
source_block = source_document.block_index[ref.block_id]
|
|
matches = list(re.finditer(re.escape(source_block.text), body))
|
|
if len(matches) != 1:
|
|
skipped.append(
|
|
f"rewrite[{index}]: source block changed before semantic replacement"
|
|
)
|
|
continue
|
|
match = matches[0]
|
|
patches.append((match.start(), match.end(), unit.replacement))
|
|
operations.append(
|
|
Operation(
|
|
type="SEMANTIC_REWRITE_BLOCK",
|
|
signal="semantic_plan",
|
|
block_id=ref.block_id,
|
|
quote=source_block.text,
|
|
replacement=unit.replacement,
|
|
source_quotes=[item.quote for item in unit.source_refs],
|
|
)
|
|
)
|
|
|
|
for start, end, replacement in sorted(patches, reverse=True):
|
|
body = body[:start] + replacement + body[end:]
|
|
|
|
if summary_values:
|
|
current = parse_document(frontmatter + body)
|
|
existing: dict[str, Any] = {}
|
|
for block in current.blocks:
|
|
if block.kind == "heading":
|
|
match = HEADING_RE.match(block.text)
|
|
key = section_key(match.group(2)) if match else None
|
|
if key:
|
|
existing[key] = block
|
|
insertions: dict[int, list[str]] = defaultdict(list)
|
|
new_sections: list[str] = []
|
|
for key in SECTION_ORDER:
|
|
values = summary_values.get(key, [])
|
|
if not values:
|
|
continue
|
|
retained: list[tuple[SemanticRewriteUnit, str]] = []
|
|
seen: set[str] = set()
|
|
existing_text = _section_text(current, existing[key]) if key in existing else ""
|
|
for unit, value in values:
|
|
if (
|
|
value in seen
|
|
or value in existing_text
|
|
or unit.replacement in existing_text
|
|
):
|
|
skipped.append(
|
|
f"add_summary[{key}]: equivalent summary already exists"
|
|
)
|
|
continue
|
|
seen.add(value)
|
|
retained.append((unit, value))
|
|
values = retained
|
|
if not values:
|
|
continue
|
|
payload = current.newline.join(value for _, value in values)
|
|
if key in existing:
|
|
insertions[existing[key].end_offset].append(
|
|
current.newline + payload + current.newline
|
|
)
|
|
else:
|
|
new_sections.append(
|
|
f"## {_heading_for(key, current.language)}"
|
|
f"{current.newline}{current.newline}{payload}"
|
|
)
|
|
for unit, value in values:
|
|
operations.append(
|
|
Operation(
|
|
type="ADD_GROUNDED_SUMMARY",
|
|
signal="semantic_plan",
|
|
target_section=key,
|
|
replacement=value,
|
|
source_quotes=[item.quote for item in unit.source_refs],
|
|
)
|
|
)
|
|
if new_sections:
|
|
first_h2 = next(
|
|
(
|
|
block
|
|
for block in current.blocks
|
|
if block.kind == "heading" and (block.heading_level or 0) >= 2
|
|
),
|
|
None,
|
|
)
|
|
if first_h2:
|
|
position = first_h2.start_offset
|
|
prefix = ""
|
|
else:
|
|
first_h1 = next(
|
|
(
|
|
block
|
|
for block in current.blocks
|
|
if block.kind == "heading" and block.heading_level == 1
|
|
),
|
|
None,
|
|
)
|
|
position = first_h1.end_offset if first_h1 else 0
|
|
prefix = current.newline
|
|
insertions[position].append(
|
|
prefix
|
|
+ (current.newline * 2).join(new_sections)
|
|
+ current.newline * 2
|
|
)
|
|
for position, values in sorted(insertions.items(), reverse=True):
|
|
body = body[:position] + "".join(values) + body[position:]
|
|
return frontmatter + body, operations, skipped
|