"""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