orchestrata_ultimate.refactor.ctok.edit_buffer

Reference › orchestrata_ultimate › refactor › ctok › edit_buffer

Byte-span edits and the safety gate that checks them. lossless_module applies an EditBuffer, and the module imports only the traceability decorator (REQ-110-124 and REQ-110-127).

EditBuffer.add(start, end, new_text) queues a replacement for a byte range and rejects a negative start or a start after the end. apply(source) first sorts the edits and refuses any span beyond the source length or overlapping its predecessor (ValueError), then splices them in reverse offset order so earlier edits do not shift later offsets; every byte outside an edit is unchanged. verify_edit(before, after, expect_equivalent=...) is the gate: both versions must parse, and if the edit claims to be formatting-only the two ASTs, rendered by stringify_ast, must be identical. If it claims to be a rename, the ASTs may differ only in identifier fields (Name.id, Attribute.attr, def and class names, arg.arg) and are compared with identifiers masked. A violation raises EditSafetyError. The comparison approach is derived from Black's AST equivalence check.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

EditSafetyError

Bases: ValueError · Decorators: traces(spec='110', story='US-110-009', reqs=['REQ-110-127']) · Line 49

Raised when an edit violates its claimed AST-equivalence contract.

EditBuffer (dataclass)

Decorators: traces(spec='110', story='US-110-009', reqs=['REQ-110-124']), dataclass(slots=True) · Line 55

Accumulates byte-span edits and applies them as splices.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/refactor/ctok/edit_buffer.py

"""EditBuffer -- span-splice edits + the Black-derived edit-safety gate.

P2 of the ctok-parser contract (plans/contracts/ctok-parser.contract.json,
REQ-110-124/127): ``EditBuffer`` accumulates byte-span edits and applies
them as splices, leaving every untouched byte identical to the original
(REQ-110-124). ``stringify_ast``/``verify_edit`` are our own small port of
Black's ``assert_equivalent`` machinery (``black/parsing.py
stringify_ast``, ``black/__init__.py assert_equivalent``) -- a
position-independent AST dump used to prove an edit didn't change program
semantics. We do NOT import Black; ``ast`` node ``_fields`` already
excludes ``lineno``/``col_offset``/``end_lineno``/``end_col_offset``
(those live in ``_attributes``), so walking only ``_fields`` gives
position-independence for free, exactly as Black's original does.
"""

from __future__ import annotations

import ast
from collections.abc import Iterator
from dataclasses import dataclass, field

from orchestrata_ultimate.core.base.traces import traces

# ast (node type, field name) pairs that hold a rename-able identifier.
# `verify_edit`'s rename mode masks these out before comparing -- any
# OTHER difference between before/after (including type changes here,
# since a masked field only fires for `str` values) fails the gate.
# Every ``ast`` field typed ``identifier`` in the Python ASDL grammar: a rename
# may legitimately touch ANY of these, so ``verify_edit`` masks them all when
# proving an edit "differs only in identifiers". Omitting a field (e.g. the
# import-name fields below) falsely rejects a valid rename that reaches an import
# statement. (``Global.names``/``Nonlocal.names`` are ``list[identifier]`` and are
# already skipped wholesale by the string dump, so they need no entry here.)
_IDENTIFIER_FIELDS: dict[type[ast.AST], frozenset[str]] = {
    ast.Name: frozenset({"id"}),
    ast.arg: frozenset({"arg"}),
    ast.FunctionDef: frozenset({"name"}),
    ast.AsyncFunctionDef: frozenset({"name"}),
    ast.ClassDef: frozenset({"name"}),
    ast.Attribute: frozenset({"attr"}),
    ast.alias: frozenset({"name", "asname"}),  # import / from-import names
    ast.ImportFrom: frozenset({"module"}),  # `from <module> import ...`
    ast.keyword: frozenset({"arg"}),  # call keyword-argument names
    ast.ExceptHandler: frozenset({"name"}),  # `except E as <name>`
}


@traces(spec="110", story="US-110-009", reqs=["REQ-110-127"])
class EditSafetyError(ValueError):
    """Raised when an edit violates its claimed AST-equivalence contract."""


@traces(spec="110", story="US-110-009", reqs=["REQ-110-124"])
@dataclass(slots=True)
class EditBuffer:
    """Accumulates byte-span edits and applies them as splices.

    Attributes:
        _edits: ``(start, end, new_text)`` triples in the order they were
            added; add order does not matter for ``apply`` (it re-sorts).
    """

    _edits: list[tuple[int, int, bytes]] = field(default_factory=list)

    @traces(spec="110", story="US-110-009", reqs=["REQ-110-124"])
    def add(self, start: int, end: int, new_text: bytes) -> None:
        """Accumulate a span edit.

        Args:
            start: Inclusive start byte offset.
            end: Exclusive end byte offset.
            new_text: Replacement bytes for ``[start, end)``.

        Raises:
            ValueError: ``start`` is negative or ``start > end``.
        """
        if start < 0 or start > end:
            raise ValueError(f"invalid edit span [{start}, {end})")
        self._edits.append((start, end, new_text))

    @traces(spec="110", story="US-110-009", reqs=["REQ-110-124"])
    def apply(self, source: bytes) -> bytes:
        """Splice all accumulated edits over ``source``.

        Edits are applied in reverse offset order so earlier splices never
        invalidate the byte offsets of later ones; every byte outside an
        edit span is byte-identical to ``source`` (REQ-110-124).

        Args:
            source: The original bytes to splice.

        Returns:
            The spliced bytes.

        Raises:
            ValueError: an edit span is out of bounds for ``source``, or
                two edit spans overlap.
        """
        ordered = sorted(self._edits, key=lambda edit: edit[0])
        prev_end = 0
        for start, end, _ in ordered:
            if end > len(source):
                raise ValueError(
                    f"edit span [{start}, {end}) exceeds {len(source)}-byte source"
                )
            if start < prev_end:
                raise ValueError(
                    f"overlapping edits: span [{start}, {end}) starts before "
                    f"the prior edit ends at {prev_end}"
                )
            prev_end = end

        buf = bytearray(source)
        for start, end, new_text in sorted(
            ordered, key=lambda edit: edit[0], reverse=True
        ):
            buf[start:end] = new_text
        return bytes(buf)


def stringify_ast(source: bytes) -> str:
    """A normalized, position-independent string dump of ``source``'s AST.

    Adapted from Black's ``black.parsing.stringify_ast`` /
    ``assert_equivalent`` (we do not import Black): walks only
    ``node._fields`` (never ``_attributes``, where ``lineno``/
    ``col_offset`` live), so two ASTs that differ only in source position
    dump identically.

    Args:
        source: Python source bytes.

    Returns:
        A multi-line string uniquely determined by the AST's shape and
        field values -- not by formatting, whitespace, or position.

    Raises:
        EditSafetyError: ``source`` does not parse as Python.
    """
    try:
        tree = ast.parse(source)
    except SyntaxError as exc:
        raise EditSafetyError(f"source failed to parse: {exc}") from exc
    return "\n".join(_stringify_ast(tree, []))


@traces(spec="110", story="US-110-009", reqs=["REQ-110-127"])
def verify_edit(before: bytes, after: bytes, *, expect_equivalent: bool) -> None:
    """The edit-safety gate: prove an edit didn't do more than it claimed.

    Args:
        before: Source bytes before the edit.
        after: Source bytes after the edit.
        expect_equivalent: If True, the edit claims to be formatting-only /
            no-semantic-change: ``before`` and ``after`` must produce an
            IDENTICAL AST (REQ-110-127). If False, the edit claims to be a
            rename: the ASTs may differ, but ONLY in identifier fields
            (``Name.id``, ``Attribute.attr``,
            ``FunctionDef``/``AsyncFunctionDef``/``ClassDef.name``,
            ``arg.arg``) -- any other structural or literal difference
            fails the gate.

    Raises:
        EditSafetyError: either source fails to parse, or the ASTs differ
            more than ``expect_equivalent`` permits.
    """
    try:
        before_tree = ast.parse(before)
    except SyntaxError as exc:
        raise EditSafetyError(f"'before' source failed to parse: {exc}") from exc
    try:
        after_tree = ast.parse(after)
    except SyntaxError as exc:
        raise EditSafetyError(f"edit produced invalid syntax: {exc}") from exc

    if expect_equivalent:
        before_dump = "\n".join(_stringify_ast(before_tree, []))
        after_dump = "\n".join(_stringify_ast(after_tree, []))
        if before_dump != after_dump:
            raise EditSafetyError(
                "edit claimed to be formatting-only / no-semantic-change but "
                "the AST changed"
            )
        return

    before_masked = "\n".join(_stringify_ast(before_tree, [], mask_identifiers=True))
    after_masked = "\n".join(_stringify_ast(after_tree, [], mask_identifiers=True))
    if before_masked != after_masked:
        raise EditSafetyError(
            "rename edit changed the AST in more than the renamed identifiers"
        )


@traces(spec="110", story="US-110-009", reqs=["REQ-110-127"])
def _stringify_ast(
    node: ast.AST, parent_stack: list[ast.AST], *, mask_identifiers: bool = False
) -> Iterator[str]:
    """Yield lines of a position-independent AST dump (see ``stringify_ast``)."""
    yield f"{'    ' * len(parent_stack)}{node.__class__.__name__}("

    for name in sorted(node._fields):
        try:
            value: object = getattr(node, name)
        except AttributeError:
            continue

        yield f"{'    ' * (len(parent_stack) + 1)}{name}="

        if isinstance(value, list):
            for item in value:
                if isinstance(item, ast.AST):
                    parent_stack.append(node)
                    yield from _stringify_ast(
                        item, parent_stack, mask_identifiers=mask_identifiers
                    )
                    parent_stack.pop()
        elif isinstance(value, ast.AST):
            parent_stack.append(node)
            yield from _stringify_ast(
                value, parent_stack, mask_identifiers=mask_identifiers
            )
            parent_stack.pop()
        else:
            masked = (
                mask_identifiers
                and isinstance(value, str)
                and name in _IDENTIFIER_FIELDS.get(type(node), frozenset())
            )
            shown: object = "<IDENTIFIER>" if masked else value
            yield (
                f"{'    ' * (len(parent_stack) + 1)}{shown!r},  # "
                f"{value.__class__.__name__}"
            )

    yield f"{'    ' * len(parent_stack)})  # /{node.__class__.__name__}"