nx_post_support/server/src/tools/incremental_parse.py

"""Reparse only the top-level commands touched by an edit.

Tcl top-level commands are independent once the previous command ended on its
own line: the parser keeps no state between them. An edit is therefore
reparsed from the first to the last top-level command it touches, commands
before it are reused as-is and commands after it are reused with their line
numbers shifted. Whenever that assumption could break, the caller falls back
to a full parse.
"""

from __future__ import annotations

import copy
from collections.abc import Callable

from tclint.syntax_tree import Node, Script
from tclint.violations import Violation

ParseChunk = Callable[[str, tuple[int, int]], tuple[Script, list[Violation]]]


def normalize_newlines(source: str) -> str:
    """Match the universal newline handling of the tclint parser."""
    return source.replace("\r\n", "\n").replace("\r", "\n")


def _shifted_tree(node: Node, delta: int) -> Node:
    """Copy a subtree with its line numbers moved by `delta`.

    Cached trees may still be read by other requests, so nodes are never
    mutated. Attributes such as `Command.routine` alias entries of `children`,
    so every reference is remapped to the same copy.
    """
    copies: dict[int, Node] = {}

    def shifted(original: Node) -> Node:
        existing = copies.get(id(original))
        if existing is not None:
            return existing
        clone = object.__new__(type(original))
        copies[id(original)] = clone
        state = dict(original.__dict__)
        if state.get("line") is not None:
            state["line"] += delta
        end = state.get("end_pos")
        if end is not None:
            state["end_pos"] = (end[0] + delta, end[1])
        for key, value in state.items():
            if isinstance(value, Node):
                state[key] = shifted(value)
            elif isinstance(value, (list, tuple)) and value and isinstance(value[0], Node):
                state[key] = type(value)(shifted(item) for item in value)
        clone.__dict__.update(state)
        return clone

    return shifted(node)


def _shifted_violation(violation: Violation, delta: int) -> Violation:
    clone = copy.copy(violation)
    clone.start = (violation.start[0] + delta, violation.start[1])
    clone.end = (violation.end[0] + delta, violation.end[1])
    return clone


def reparse(
    old_source: str,
    old_tree: Script,
    old_violations: list[Violation],
    new_source: str,
    parse_chunk: ParseChunk,
) -> tuple[Script, list[Violation]] | None:
    """Return the tree of `new_source`, or None when a full parse is needed.

    Both sources must already be newline-normalized. `parse_chunk` parses a
    top-level fragment starting at the given (line, column) and may raise
    TclSyntaxError, which the caller handles like any failed parse.
    """
    if old_source == new_source:
        return old_tree, list(old_violations)

    # Changed line range (1-indexed); lines outside it are identical. A pure
    # insertion leaves last_changed_old == first_changed_line - 1.
    old_lines = old_source.split("\n")
    new_lines = new_source.split("\n")
    limit = min(len(old_lines), len(new_lines))
    same_before = 0
    while same_before < limit and old_lines[same_before] == new_lines[same_before]:
        same_before += 1
    same_after = 0
    while (
        same_after < limit - same_before
        and old_lines[-1 - same_after] == new_lines[-1 - same_after]
    ):
        same_after += 1

    first_changed_line = same_before + 1
    last_changed_old = len(old_lines) - same_after
    delta = len(new_lines) - len(old_lines)

    commands = old_tree.children
    if any(command.line is None or command.end_pos is None for command in commands):
        return None

    # Commands overlapping the changed lines, widened so that no reused
    # command shares a line with the reparsed range.
    first = next(
        (index for index, command in enumerate(commands) if command.end_pos[0] >= first_changed_line),
        len(commands),
    )
    start_line = first_changed_line
    if first < len(commands):
        start_line = min(start_line, commands[first].line)
    while first > 0 and commands[first - 1].end_pos[0] >= start_line:
        first -= 1
        start_line = min(start_line, commands[first].line)

    last = first - 1
    end_line_old = last_changed_old
    while last + 1 < len(commands) and commands[last + 1].line <= end_line_old:
        last += 1
        end_line_old = max(end_line_old, commands[last].end_pos[0])
    end_line_new = end_line_old + delta

    if end_line_new < start_line - 1 or end_line_new > len(new_lines):
        return None
    # A trailing backslash joins a line with the next one across the boundary.
    if start_line > 1 and new_lines[start_line - 2].endswith("\\"):
        return None
    if end_line_new >= start_line and new_lines[end_line_new - 1].endswith("\\"):
        return None

    chunk_commands: list[Node] = []
    chunk_violations: list[Violation] = []
    if end_line_new >= start_line:
        chunk = "\n".join(new_lines[start_line - 1 : end_line_new])
        chunk_tree, chunk_violations = parse_chunk(chunk, (start_line, 1))
        chunk_commands = chunk_tree.children

    reused_after = [_shifted_tree(command, delta) for command in commands[last + 1 :]]
    tree = Script(
        *commands[:first],
        *chunk_commands,
        *reused_after,
        pos=(old_tree.line, old_tree.col),
    )
    tree.end_pos = (len(new_lines), len(new_lines[-1]) + 1)

    violations = [violation for violation in old_violations if violation.start[0] < start_line]
    violations += chunk_violations
    violations += [
        _shifted_violation(violation, delta)
        for violation in old_violations
        if violation.start[0] > end_line_old
    ]
    return tree, violations