textmachine/eval/editor_contract/apply.py

#!/usr/bin/env python3
"""Толерантный детерминированный apply anchored search/replace — Python-порт Go-стиля,
по спеке `docs/research/19-chunking-cohesion.md` §C2 (риг арма Q4b, `exp15:108`).

Спека дословно: якорь ищется с нормализацией пробелов/пунктуационных вариантов; неоднозначный
якорь (≥2 совпадений) → REJECT операции (не тихое применение первого); якорь не найден → REJECT;
reject-rate — телеметрия. Малформед → 1 регенерация → далее операция ОТБРАСЫВАЕТСЯ (черновик
остаётся) — blast-radius одной правки, а не чанка.
"""
from __future__ import annotations
import re
import unicodedata
from dataclasses import dataclass, field

# ⚠ ФИКС-2 (адверсариальная приёмка 05.08): прежняя форма требовала `\n` ПЕРЕД закрывающим маркером
# и потому МОЛЧА теряла операцию с ПУСТОЙ заменой (удаление фрагмента) — она не попадала ни в
# ops_total, ни в malformed. Формат удаление выражает штатно; терял его парсер.
OP_RE = re.compile(
    r"<{5,9}[ \t]*SEARCH[ \t]*\n(?P<search>.*?)\n={5,9}[ \t]*\n(?P<replace>.*?)\n?>{5,9}[ \t]*REPLACE",
    re.S)
NO_CHANGE = "NO_CHANGE"

# Пунктуационные варианты, которые модель законно путает при копировании якоря.
PUNCT_CLASSES = [
    ("«»\"“”„‟", '"'),
    ("—–‒−-", "-"),
    ("…", "..."),
    ("’‘'`", "'"),
]


def _fold_chars(text: str):
    """Посимвольная свёртка ОРИГИНАЛЬНОГО текста → [(символ_свёртки, индекс_в_ОРИГИНАЛЕ)].

    ⚠ ФИКС-1 (адверсариальная приёмка 05.08, критический). Прежняя версия строила карту офсетов по
    `unicodedata.normalize("NFKC", text)`, а РЕЗАЛА оригинал: символ, меняющий длину при
    нормализации («…» → «...»), сдвигал все последующие офсеты, и apply вставлял замену не туда,
    молча склеивая текст («Он ждал… потом ушёл. НеНебо было серым»). Поймано на 6 из 36 выходов
    (окно w4), при этом риг рапортовал applied=ops_total и rejected=0.
    Лечение: нормализация ПОСИМВОЛЬНАЯ, индекс всегда указывает в ОРИГИНАЛ; расширение одного
    символа в несколько даёт несколько записей с ОДНИМ и тем же исходным индексом.
    """
    out = []
    prev_space = True                       # схлоп ведущих пробелов
    for i, ch in enumerate(text):
        if ch.isspace():
            if not prev_space:
                out.append((" ", i))
                prev_space = True
            continue
        prev_space = False
        norm_ch = unicodedata.normalize("NFKC", ch)
        for chars, to in PUNCT_CLASSES:
            if ch in chars:
                norm_ch = to
                break
        for c in norm_ch:
            out.append((c, i))
    return out


def fold(s: str) -> str:
    """Нормализация ДЛЯ ПОИСКА (не для вывода). Тот же путь, что у _fold_map, — иначе ключ и стог
    могли бы свернуться по-разному (полнострочный NFKC компонует то, чего посимвольный не делает)."""
    return "".join(c for c, _ in _fold_chars(s)).strip()


def _fold_map(text: str) -> tuple[str, list[int]]:
    """Свёрнутый текст + карта «индекс в свёрнутом → индекс в ОРИГИНАЛЕ»."""
    pairs = _fold_chars(text)
    return "".join(c for c, _ in pairs), [i for _, i in pairs]


@dataclass
class ApplyResult:
    text: str
    ops_total: int = 0
    applied: int = 0
    rejected_notfound: int = 0
    rejected_ambiguous: int = 0
    rejected_overlap: int = 0
    rejected_empty: int = 0
    malformed: bool = False
    no_change: bool = False
    details: list[str] = field(default_factory=list)

    @property
    def rejected(self) -> int:
        return (self.rejected_notfound + self.rejected_ambiguous
                + self.rejected_overlap + self.rejected_empty)

    @property
    def compliance(self) -> float:
        """Доля операций, доехавших до применения. Стоп-гейт Q4b: ≥0.90."""
        return self.applied / self.ops_total if self.ops_total else 1.0


def parse_ops(reply: str) -> tuple[list[tuple[str, str]], bool]:
    """→ ([(search, replace)], malformed). NO_CHANGE = пустой список, не малформед."""
    body = reply.strip()
    ops = [(m.group("search"), m.group("replace")) for m in OP_RE.finditer(body)]
    if ops:
        # мусор ВНЕ операций (преамбулы/комментарии) — нарушение правила 6, но операции валидны:
        # считаем малформедом только если разметка вообще не распозналась
        return ops, False
    if NO_CHANGE in body and len(body) < 200:
        return [], False
    return [], True


def apply_ops(draft: str, reply: str) -> ApplyResult:
    ops, malformed = parse_ops(reply)
    res = ApplyResult(text=draft, malformed=malformed, no_change=(not ops and not malformed))
    if malformed or not ops:
        return res
    res.ops_total = len(ops)
    folded, idx = _fold_map(draft)
    spans: list[tuple[int, int, str]] = []
    for search, replace in ops:
        key = fold(search)
        if not key:
            res.rejected_empty += 1
            res.details.append("REJECT пустой якорь")
            continue
        hits = [m.start() for m in re.finditer(re.escape(key), folded)]
        if not hits:
            res.rejected_notfound += 1
            res.details.append(f"REJECT якорь не найден: {search[:60]!r}")
            continue
        if len(hits) > 1:
            res.rejected_ambiguous += 1
            res.details.append(f"REJECT якорь неоднозначен ({len(hits)}): {search[:60]!r}")
            continue
        s = idx[hits[0]]
        e_fold = hits[0] + len(key) - 1
        e = idx[e_fold] + 1
        spans.append((s, e, replace))
    # пересечения: операция, перекрывающая уже принятую, отбрасывается (спека: не пересекаются)
    spans.sort()
    accepted: list[tuple[int, int, str]] = []
    for s, e, r in spans:
        if accepted and s < accepted[-1][1]:
            res.rejected_overlap += 1
            res.details.append(f"REJECT пересечение спанов на {s}")
            continue
        accepted.append((s, e, r))
    out = draft
    for s, e, r in sorted(accepted, reverse=True):   # с хвоста — офсеты не едут
        out = out[:s] + r + out[e:]
    res.text = out
    res.applied = len(accepted)
    return res