From b5c9ff1345a451956fc47ccfd72e31ed49e234cd Mon Sep 17 00:00:00 2001 From: vitali87 Date: Sun, 20 Sep 2026 22:56:51 +0100 Subject: [PATCH] feat: plan change-signature graph edits --- codebase_rag/constants/ast_nodes.py | 1 + codebase_rag/constants/ast_python.py | 3 + codebase_rag/constants/cli.py | 45 ++++ codebase_rag/constants/graph.py | 2 + codebase_rag/constants/mcp.py | 2 + codebase_rag/cypher_queries.py | 6 + codebase_rag/editing/contract.py | 5 +- codebase_rag/editing/signature_calls.py | 136 ++++++++++++ codebase_rag/editing/signature_literals.py | 55 +++++ codebase_rag/editing/signature_params.py | 246 +++++++++++++++++++++ codebase_rag/editing/signature_planner.py | 241 ++++++++++++++++++++ codebase_rag/editing/signature_rewrite.py | 139 ++++++++++++ codebase_rag/editing/signature_types.py | 103 +++++++++ evals/cgr_graph.py | 14 ++ 14 files changed, 997 insertions(+), 1 deletion(-) create mode 100644 codebase_rag/editing/signature_calls.py create mode 100644 codebase_rag/editing/signature_literals.py create mode 100644 codebase_rag/editing/signature_params.py create mode 100644 codebase_rag/editing/signature_planner.py create mode 100644 codebase_rag/editing/signature_rewrite.py create mode 100644 codebase_rag/editing/signature_types.py diff --git a/codebase_rag/constants/ast_nodes.py b/codebase_rag/constants/ast_nodes.py index 3803a1f5d8..d4da1840ef 100644 --- a/codebase_rag/constants/ast_nodes.py +++ b/codebase_rag/constants/ast_nodes.py @@ -195,6 +195,7 @@ # tree-sitter parses comments as named children, so the flow walk filters them # out before indexing arguments or reading a single sub-expression. TS_COMMENT = "comment" +TS_SPREAD_ELEMENT = "spread_element" # `(expr)` wraps its value in a parenthesized_expression; the flow walk unwraps # it (like await) to reach the inner source/tainted expression. TS_PARENTHESIZED_EXPRESSION = "parenthesized_expression" diff --git a/codebase_rag/constants/ast_python.py b/codebase_rag/constants/ast_python.py index 87f52e0447..4b19eee217 100644 --- a/codebase_rag/constants/ast_python.py +++ b/codebase_rag/constants/ast_python.py @@ -49,6 +49,9 @@ TS_PY_DEFAULT_PARAMETER = "default_parameter" TS_PY_LIST_SPLAT_PATTERN = "list_splat_pattern" TS_PY_DICTIONARY_SPLAT_PATTERN = "dictionary_splat_pattern" +# Call-site splats (`*args`, `**kwargs` in an argument list). +TS_PY_LIST_SPLAT = "list_splat" +TS_PY_DICTIONARY_SPLAT = "dictionary_splat" TS_PY_POSITIONAL_SEPARATOR = "positional_separator" TS_PY_KEYWORD_SEPARATOR = "keyword_separator" TS_PY_SUBSCRIPT = "subscript" diff --git a/codebase_rag/constants/cli.py b/codebase_rag/constants/cli.py index 185e4669ed..07cafd39f5 100644 --- a/codebase_rag/constants/cli.py +++ b/codebase_rag/constants/cli.py @@ -605,6 +605,51 @@ class DiffMarker: RENAME_CONTRACT_UNMEASURED = ( "Rename applied, but its postcondition could not be measured: {error}" ) +# change_signature (issue #1533). +SIGNATURE_BAD_SPEC = ( + "Not a parameter spec: {spec!r} (expected name[:annotation][=default][@source])" +) +SIGNATURE_UNKNOWN = "No function or method named {qn} in the graph" +SIGNATURE_NO_GRAMMAR = "No grammar for {path}; the signature cannot be rewritten" +SIGNATURE_NO_DEFINITION_TOKEN = "Could not locate the definition of {qn} in {path}" +SIGNATURE_DUPLICATE_PARAM = "Parameter {name} is listed twice" +SIGNATURE_POSITIONAL_ONLY_MOVED = ( + "The definition marks its leading parameters positional-only with `/`, " + "and this rewrite moves a parameter across that boundary. Rebuilding it " + "would change which arguments callers may pass by name. Rewrite the " + "parameters before `/` in place, or remove the marker first." +) +SIGNATURE_DUPLICATE_SOURCE = ( + "Parameters {names} all map to {source}, which holds one value per " + "call site: mapping it to several would leave every caller short of " + "arguments. Map each new parameter to its own source, or give the " + "extras a literal default." +) +SIGNATURE_UNKNOWN_SOURCE = ( + "No old parameter {source!r}; the old parameters are: {names}" +) +SIGNATURE_LITERAL_TYPE = ( + "Default {literal} for {name} does not fit its declared type {annotation}" +) +SIGNATURE_PLANNED = "{count} call site(s) would be rewritten, {unmapped} left unmapped" +SIGNATURE_PARSE_FAILED = "Signature change rolled back: {files} would no longer parse" +SIGNATURE_CONTRACT_FAILED = ( + "Signature change rolled back, postcondition failed: {reasons}" +) +SIGNATURE_UNLOCATABLE = "site cannot be located" +SIGNATURE_GUESSED = "resolved by {resolution}; pass allow_heuristic to rewrite" +SIGNATURE_MISSING_FILE = "file is missing" +SIGNATURE_NO_CALL = "no call with an argument list at the recorded site" +SIGNATURE_SPLAT = "the site spreads its arguments" +SIGNATURE_VARIADIC = ( + "Cannot change the signature of {qn}: it takes *args or **kwargs, so its call " + "sites cannot be mapped one value per parameter" +) +SIGNATURE_UNKNOWN_KEYWORD = "the site passes an unknown keyword {name}" +SIGNATURE_NEEDS_KEYWORDS = "the mapping needs keyword arguments this language lacks" +SIGNATURE_UNMAPPED_PARAM = "the site passes no value for {name}" +SIGNATURE_SURPLUS_ARGS = "the site passes more arguments than the definition declares" +SIGNATURE_DEFAULT_ORDER = "Parameter {name} has no default but follows one that does" # Postcondition contract (issue #1531). CONTRACT_OP_RENAME = "rename" CONTRACT_OP_CHANGE_SIGNATURE = "change_signature" diff --git a/codebase_rag/constants/graph.py b/codebase_rag/constants/graph.py index 9c81b08374..4c66bb740a 100644 --- a/codebase_rag/constants/graph.py +++ b/codebase_rag/constants/graph.py @@ -55,6 +55,8 @@ KEY_SITES = "sites" KEY_AMBIGUOUS = "ambiguous" KEY_STRUCTURAL_DELTA = "structural_delta" +KEY_UNMAPPED = "unmapped" +KEY_VERDICT = "verdict" KEY_DISPATCH_LITERAL = "dispatch_literal" diff --git a/codebase_rag/constants/mcp.py b/codebase_rag/constants/mcp.py index 364957ed94..242d3930c7 100644 --- a/codebase_rag/constants/mcp.py +++ b/codebase_rag/constants/mcp.py @@ -21,6 +21,7 @@ class MCPToolName(StrEnum): TESTS_REACHING = "tests_reaching" # Graph-driven edit operations (issue #1532). RENAME = "rename" + CHANGE_SIGNATURE = "change_signature" QUERY_CODE_GRAPH = "query_code_graph" GET_CODE_SNIPPET = "get_code_snippet" SURGICAL_REPLACE_CODE = "surgical_replace_code" @@ -102,6 +103,7 @@ class MCPParamName(StrEnum): MODULE_QN = "module_qualified_name" NEW_NAME = "new_name" ALLOW_HEURISTIC = "allow_heuristic" + NEW_PARAMS = "new_params" # MCP server constants diff --git a/codebase_rag/cypher_queries.py b/codebase_rag/cypher_queries.py index f89d910c4b..fc6c86d6d5 100644 --- a/codebase_rag/cypher_queries.py +++ b/codebase_rag/cypher_queries.py @@ -575,6 +575,12 @@ def build_create_relationship_query( AND t.qualified_name STARTS WITH $project_prefix RETURN DISTINCT m.qualified_name AS from_qn, m.path AS from_path, t.qualified_name AS to_qn""" +# The declared parameter shape of one definition (issues #1527, #1533). +CYPHER_GRAPH_SIGNATURE = """MATCH (n:Function|Method) +WHERE n.qualified_name = $qn AND n.qualified_name STARTS WITH $project_prefix +RETURN n.positional_params AS positional_params, n.param_types AS param_types, + n.return_type AS return_type +LIMIT 1""" CYPHER_GRAPH_IMPORTERS = """MATCH (m:Module)-[r:IMPORTS]->(target) WHERE target.qualified_name = $qn AND m.qualified_name STARTS WITH $project_prefix RETURN m.qualified_name AS qualified_name, m.path AS path, r.line AS line, diff --git a/codebase_rag/editing/contract.py b/codebase_rag/editing/contract.py index 0e3b974499..a04c3db93d 100644 --- a/codebase_rag/editing/contract.py +++ b/codebase_rag/editing/contract.py @@ -86,12 +86,15 @@ def rename_expectation( ) -def change_signature_expectation(unmapped: Iterable[str]) -> Expectation: +def change_signature_expectation( + unmapped: Iterable[str], heuristic_allowed: bool = False +) -> Expectation: return Expectation( operation=cs.CONTRACT_OP_CHANGE_SIGNATURE, unmapped=tuple(sorted(set(unmapped))), # A parameter added with a default changes no caller count. caller_count_unchanged=True, + heuristic_allowed=heuristic_allowed, ) diff --git a/codebase_rag/editing/signature_calls.py b/codebase_rag/editing/signature_calls.py new file mode 100644 index 0000000000..8a9e6b2d1a --- /dev/null +++ b/codebase_rag/editing/signature_calls.py @@ -0,0 +1,136 @@ +"""Call-site argument mapping for change_signature.""" + +from __future__ import annotations + +from tree_sitter import Node + +from .. import constants as cs +from .. import graph_query +from ..parsers.call_processor import _find_call_arguments_node, _split_call_arguments +from ..types_defs import ResultRow +from .signature_params import _text +from .signature_types import ParamSpec + +_SPLAT_TYPES = frozenset( + { + cs.TS_PY_LIST_SPLAT, + cs.TS_PY_DICTIONARY_SPLAT, + cs.TS_SPREAD_ELEMENT, + } +) + + +def _site_end(row: graph_query.CallSiteRow | ResultRow) -> tuple[int, int] | None: + """The site's recorded end point as tree-sitter counts it, if recorded.""" + end_line, end_col = row.get("end_line"), row.get("end_col") + if isinstance(end_line, int) and isinstance(end_col, int): + return (end_line - 1, end_col) + return None + + +def _call_at( + root: Node, line: int, col: int, end: tuple[int, int] | None = None +) -> Node | None: + """The call node at (line, col) that carries arguments. + + With the site's recorded end point the exact call is chosen, so a + chained `helper(2).upper()` rewrites `helper`'s arguments and not the + outer call's; without it the outermost call at that point is taken. + """ + stack = [root] + found: Node | None = None + while stack: + node = stack.pop() + if ( + node.start_point == (line - 1, col) + and _find_call_arguments_node(node) is not None + ): + if end is not None: + if node.end_point == end: + return node + elif found is None or node.end_byte > found.end_byte: + found = node + if node.start_point[0] <= line - 1 <= node.end_point[0]: + stack.extend(node.children) + return found + + +def _map_arguments( + args_node: Node, + old_names: list[str], + specs: list[ParamSpec], + language: cs.SupportedLanguage | None, + keyword_only: frozenset[str] = frozenset(), +) -> list[str] | str: + """The site's new argument texts, or the reason it cannot be mapped. + + A keyword naming a keyword-only parameter is not part of the positional + mapping and is carried through exactly as written. + """ + positional, keyword = _split_call_arguments(args_node) + if any(child.type in _SPLAT_TYPES for child in args_node.named_children): + return cs.SIGNATURE_SPLAT + values: dict[int, tuple[str, bool]] = {} + for index, node in enumerate(positional): + values[index] = (_text(node), False) + carried: list[str] = [] + for name, node in keyword.items(): + if name in keyword_only: + carried.append(f"{name}={_text(node)}") + continue + if name not in old_names: + return cs.SIGNATURE_UNKNOWN_KEYWORD.format(name=name) + values[old_names.index(name)] = (_text(node), True) + keywords_ok = language == cs.SupportedLanguage.PYTHON + if not positional and _keywords_already_fit(values, old_names, specs): + # A keyword-only site binds by name: reordering would change + # nothing but the spelling, so it is left exactly as written. + return [_text(child) for child in args_node.named_children] + out: list[str] = [] + keyword_mode = False + for spec in specs: + if spec.from_index is not None: + found = values.pop(spec.from_index, None) + if found is None: + # The site relied on the old default; every later value + # must be spelled by name so positions do not shift. + keyword_mode = True + continue + text, was_keyword = found + if was_keyword or keyword_mode: + if not keywords_ok: + return cs.SIGNATURE_NEEDS_KEYWORDS + out.append(f"{spec.name}={text}") + keyword_mode = True + else: + out.append(text) + elif spec.literal is not None: + if keyword_mode: + if not keywords_ok: + return cs.SIGNATURE_NEEDS_KEYWORDS + out.append(f"{spec.name}={spec.literal}") + else: + out.append(spec.literal) + else: + return cs.SIGNATURE_UNMAPPED_PARAM.format(name=spec.name) + # Every spec consumed what it named; anything still in `values` is an + # argument the new signature has no home for. Rewriting would drop it + # from the caller's source silently -- and the contract cannot catch + # that, because the argument is gone from the file before the delta is + # measured, so the `too_many` arity check sees nothing. + if values: + return cs.SIGNATURE_SURPLUS_ARGS + return out + carried + + +def _keywords_already_fit( + values: dict[int, tuple[str, bool]], old_names: list[str], specs: list[ParamSpec] +) -> bool: + """Every passed keyword keeps its name and nothing new must be inserted.""" + kept = {spec.from_index: spec.name for spec in specs if spec.from_index is not None} + if any(index not in kept or kept[index] != old_names[index] for index in values): + return False + return all(spec.from_index is not None for spec in specs) + + +# --- literal type checks -------------------------------------------------------- diff --git a/codebase_rag/editing/signature_literals.py b/codebase_rag/editing/signature_literals.py new file mode 100644 index 0000000000..327238258f --- /dev/null +++ b/codebase_rag/editing/signature_literals.py @@ -0,0 +1,55 @@ +"""Literal/type compatibility checks for change_signature.""" + +from __future__ import annotations + +import ast +import re + +_KIND_OF = { + bool: "bool", + int: "int", + float: "float", + str: "str", + list: "list", + dict: "dict", + set: "set", + tuple: "tuple", + type(None): "None", +} +_ACCEPTS: dict[str, frozenset[str]] = { + "int": frozenset({"int"}), + "float": frozenset({"int", "float"}), + "str": frozenset({"str"}), + "bool": frozenset({"bool"}), + "list": frozenset({"list"}), + "dict": frozenset({"dict"}), + "set": frozenset({"set"}), + "tuple": frozenset({"tuple"}), + "bytes": frozenset(), +} + + +def _literal_kind(literal: str) -> str | None: + try: + value = ast.literal_eval(literal) + except (ValueError, SyntaxError): + return None + return _KIND_OF.get(type(value)) + + +def _compatible(kind: str, annotation: str) -> bool: + """Whether a literal of `kind` fits `annotation` (builtin names only).""" + text = annotation.strip() + if kind == "None": + return "None" in text or text.startswith("Optional") + members = [m.strip() for m in re.split(r"\s*\|\s*", text)] + if len(members) > 1: + return any(_compatible(kind, member) for member in members) + base = re.match(r"^([A-Za-z_][A-Za-z0-9_]*)", text) + if base is None: + return True + accepted = _ACCEPTS.get(base.group(1)) + if accepted is None: + # Not a builtin: the graph cannot tell, so do not accuse. + return True + return kind in accepted diff --git a/codebase_rag/editing/signature_params.py b/codebase_rag/editing/signature_params.py new file mode 100644 index 0000000000..290acd6b3b --- /dev/null +++ b/codebase_rag/editing/signature_params.py @@ -0,0 +1,246 @@ +"""Definition-parameter parsing and rendering for change_signature.""" + +from __future__ import annotations + +import re +from typing import NamedTuple + +from tree_sitter import Node + +from .. import constants as cs +from .signature_types import ParamSpec, SignatureRefused + +_RECEIVERS = frozenset({cs.PY_KEYWORD_SELF, cs.PY_KEYWORD_CLS}) +_COMMENT_TYPES = frozenset({cs.TS_COMMENT}) +_SEPARATOR_TYPES = frozenset( + {cs.TS_PY_POSITIONAL_SEPARATOR, cs.TS_PY_KEYWORD_SEPARATOR} +) +_VARIADIC_PARAM_TYPES = frozenset( + {cs.TS_PY_LIST_SPLAT_PATTERN, cs.TS_PY_DICTIONARY_SPLAT_PATTERN} +) + + +class _Param(NamedTuple): + name: str + node: Node + receiver: bool + # After a bare `*`: bound by keyword only, so it has no positional index + # for a spec to map from and stays exactly as written. + keyword_only: bool = False + + +def _text(node: Node | None) -> str: + if node is None or node.text is None: + return "" + return node.text.decode(cs.ENCODING_UTF8, errors="replace") + + +def _identifier_in(node: Node) -> str: + """The first identifier token inside a parameter node, any grammar.""" + named = node.child_by_field_name(cs.FIELD_NAME) + if named is not None and named.type in (cs.TS_IDENTIFIER, cs.TS_PY_IDENTIFIER): + return _text(named) + pattern = node.child_by_field_name(cs.TS_FIELD_PATTERN) + if pattern is not None and pattern.type == cs.TS_IDENTIFIER: + return _text(pattern) + if node.type in (cs.TS_IDENTIFIER, cs.TS_PY_IDENTIFIER): + return _text(node) + stack = list(reversed(node.children)) + while stack: + child = stack.pop() + if child.type in (cs.TS_IDENTIFIER, cs.TS_PY_IDENTIFIER): + return _text(child) + stack.extend(reversed(child.children)) + return "" + + +def _definition_node(root: Node, line: int, col: int) -> Node | None: + """The definition whose own name token starts at (line, col).""" + stack = [root] + while stack: + node = stack.pop() + named = node.child_by_field_name(cs.FIELD_NAME) + if ( + named is not None + and named.start_point == (line - 1, col) + and node.child_by_field_name(cs.FIELD_PARAMETERS) is not None + ): + return node + if node.start_point[0] <= line - 1 <= node.end_point[0]: + stack.extend(node.children) + return None + + +def _positional_only_count(definition: Node) -> int | None: + """How many leading parameters `/` makes positional-only, if it is there. + + `/` neither takes a value nor shifts the positional indices callers map + through, so `_parameters` is right to leave it out of the mapping model. + It is still load-bearing when the signature is REBUILT: dropped, the + parameters before it become keyword-callable, and a call the original + rejected starts being accepted. The count is what `_rewrite_definition` + needs to put it back. + """ + params = definition.child_by_field_name(cs.FIELD_PARAMETERS) + if params is None: + return None + seen = 0 + for child in params.named_children: + if child.type == cs.TS_PY_POSITIONAL_SEPARATOR: + return seen + if child.type in _COMMENT_TYPES or child.type in _SEPARATOR_TYPES: + continue + seen += 1 + return None + + +def _parameters( + definition: Node, language: cs.SupportedLanguage | None +) -> list[_Param]: + params = definition.child_by_field_name(cs.FIELD_PARAMETERS) + if params is None: + return [] + out: list[_Param] = [] + keyword_only = False + for child in params.named_children: + if child.type == cs.TS_PY_KEYWORD_SEPARATOR: + keyword_only = True + continue + if child.type in _COMMENT_TYPES or child.type in _SEPARATOR_TYPES: + # `/` is a marker, not a parameter: it neither takes a value nor + # shifts the positional indices callers map through. It is put + # back on rebuild from `_positional_only_count`, which is where + # it matters. + continue + name = _identifier_in(child) + receiver = ( + language == cs.SupportedLanguage.PYTHON and not out and name in _RECEIVERS + ) or child.type == cs.TS_RS_SELF_PARAMETER + out.append(_Param(name, child, receiver, keyword_only)) + return out + + +def _refuse_duplicate_sources(resolved: list[ParamSpec]) -> None: + """Refuse two new parameters fed by one old one. + + A call site holds a single value at each old position, and `_map_arguments` + consumes it: the first mapping takes the value and every later one reads as + omitted, so the rewritten definition gains a parameter that no caller + passes. The edit reports success and the callers raise `TypeError`. + + Refusing rather than emitting the value twice follows the rest of this + layer: an ambiguous request is answered with a message, not with a + plausible guess the caller has to discover at runtime. + """ + by_source: dict[int, list[str]] = {} + for spec in resolved: + if spec.from_index is not None: + by_source.setdefault(spec.from_index, []).append(spec.name) + for source, names in by_source.items(): + if len(names) > 1: + raise SignatureRefused( + cs.SIGNATURE_DUPLICATE_SOURCE.format( + names=", ".join(names), source=f"position {source}" + ) + ) + + +def _restore_positional_only( + rendered: list[str], + node: Node, + old: list[_Param], + specs: list[ParamSpec], +) -> None: + """Put `/` back, or refuse if the rewrite moved a parameter across it. + + The separator says the parameters before it cannot be passed by name. + Dropping it widens the callable contract silently: `def f(a, b, /, c)` + rebuilt as `def f(a, b, c)` accepts `f(a=1, b=2, c=3)`, which the original + rejects. So it has to come back. + + It can only come back where it still means the same thing. The marker + counts LEADING parameters, so it survives a rewrite that leaves the same + sources, in the same order, in front of it. A rewrite that reorders across + the boundary, or maps a new parameter into that region, changes which + arguments callers may name -- a question this operation is not asked to + answer, so it refuses rather than guessing. + """ + boundary = _positional_only_count(node) + if boundary is None: + return + receivers = sum(1 for p in old if p.receiver) + kept = specs[:boundary] + if len(specs) < boundary or any( + spec.from_index != index for index, spec in enumerate(kept) + ): + raise SignatureRefused(cs.SIGNATURE_POSITIONAL_ONLY_MOVED) + rendered.insert(receivers + boundary, "/") + + +def _render_param(spec: ParamSpec, language: cs.SupportedLanguage | None) -> str: + """A brand-new parameter in the target grammar's spelling.""" + annotation = spec.annotation + default = spec.default if spec.default is not None else spec.literal + if language == cs.SupportedLanguage.GO: + return f"{spec.name} {annotation}" if annotation else spec.name + if language == cs.SupportedLanguage.JAVA: + return f"{annotation} {spec.name}" if annotation else spec.name + if language == cs.SupportedLanguage.RUST: + return f"{spec.name}: {annotation}" if annotation else spec.name + text = spec.name + if annotation: + text += f": {annotation}" + if default is not None: + text += f" = {default}" if annotation else f"={default}" + return text + + +def _rendered_old( + param: _Param, spec: ParamSpec, language: cs.SupportedLanguage | None +) -> str: + """An old parameter carried into the new signature, renamed if asked.""" + text = _text(param.node) + if spec.name != param.name: + text = re.sub(rf"\b{re.escape(param.name)}\b", spec.name, text, count=1) + if spec.annotation is None and spec.default is None: + return text + # A spec that gives only one half keeps the other half from the old + # parameter: `b:str@1` over `b='x'` is `b: str = 'x'`, not `b: str`. + old_annotation, old_default = _split_old(param.node) + merged = spec._replace( + annotation=spec.annotation if spec.annotation is not None else old_annotation, + default=spec.default if spec.default is not None else old_default, + ) + return _render_param(merged, language) + + +def _split_old(node: Node) -> tuple[str | None, str | None]: + """The old parameter's annotation and default text, if it has them.""" + annotation = node.child_by_field_name(cs.FIELD_TYPE) + default = node.child_by_field_name(cs.FIELD_VALUE) + return ( + _text(annotation) if annotation is not None else None, + _text(default) if default is not None else None, + ) + + +_PY_DEFAULTED = frozenset( + {cs.TS_PY_DEFAULT_PARAMETER, cs.TS_PY_TYPED_DEFAULT_PARAMETER} +) + + +def _check_default_order(positional: list[_Param], specs: list[ParamSpec]) -> None: + """Python: no required parameter may follow one with a default.""" + seen_default = False + for spec in specs: + if spec.from_index is not None: + node = positional[spec.from_index].node + has_default = spec.default is not None or node.type in _PY_DEFAULTED + else: + has_default = spec.default is not None or spec.literal is not None + if seen_default and not has_default: + raise SignatureRefused(cs.SIGNATURE_DEFAULT_ORDER.format(name=spec.name)) + seen_default = seen_default or has_default + + +# --- call-site mapping ---------------------------------------------------------- diff --git a/codebase_rag/editing/signature_planner.py b/codebase_rag/editing/signature_planner.py new file mode 100644 index 0000000000..4a151a06c4 --- /dev/null +++ b/codebase_rag/editing/signature_planner.py @@ -0,0 +1,241 @@ +"""Planning layer for change_signature.""" + +from __future__ import annotations + +from collections.abc import Callable, Iterable +from pathlib import Path + +from tree_sitter import Node + +from .. import constants as cs +from .. import cypher_queries as cq +from .. import graph_query +from ..graph_query import QueryFn +from ..language_spec import get_language_for_extension +from ..parser_loader import load_parsers +from .contract import Reingest +from .patcher import Patcher +from .rename import _hierarchy, _name_token +from .signature_literals import _compatible, _literal_kind +from .signature_params import ( + _RECEIVERS, + _VARIADIC_PARAM_TYPES, + _definition_node, + _Param, + _parameters, + _refuse_duplicate_sources, +) +from .signature_rewrite import rewrite_definition, rewrite_site +from .signature_types import ( + ParamSpec, + RewrittenSite, + SignatureRefused, + SignatureReport, + UnmappedSite, +) +from .transaction import StagedTree, VerificationResult + + +class SignaturePlanner: + def __init__( + self, + repo_root: Path, + fetch_all: QueryFn, + project_name: str, + verify: Callable[[StagedTree], VerificationResult | bool | None] | None = None, + reingest: Reingest | None = None, + ) -> None: + self.repo_root = repo_root.resolve() + self.fetch_all = fetch_all + self.project = project_name + self.verify = verify + self.reingest = reingest + self._parsers = load_parsers()[0] + self._allow_heuristic = False + self._keyword_only: frozenset[str] = frozenset() + + def _parse( + self, path: str, source: bytes + ) -> tuple[cs.SupportedLanguage | None, Node | None]: + language = get_language_for_extension(Path(path).suffix) + parser = self._parsers.get(language) if language is not None else None + if parser is None: + return language, None + return language, parser.parse(source).root_node + + def _definition( + self, qn: str, patcher: Patcher + ) -> tuple[str, Node, cs.SupportedLanguage | None, bytes]: + row = graph_query.definition(self.fetch_all, self.project, qn, self.repo_root) + if not row["found"] or not row["path"]: + raise SignatureRefused(cs.SIGNATURE_UNKNOWN.format(qn=qn)) + path = row["path"] + source = patcher.source(path) + language, root = self._parse(path, source) + if root is None: + raise SignatureRefused(cs.SIGNATURE_NO_GRAMMAR.format(path=path)) + name = row["name"] or qn.rsplit(cs.SEPARATOR_DOT, 1)[-1] + token = _name_token( + source, + language, + row["start_line"] or 1, + row["end_line"] or 1, + name.split("(")[0], + ) + node = _definition_node(root, *token) if token else None + if node is None: + raise SignatureRefused( + cs.SIGNATURE_NO_DEFINITION_TOKEN.format(qn=qn, path=path) + ) + return path, node, language, source + + def _resolve_specs( + self, specs: Iterable[ParamSpec], old: list[_Param] + ) -> list[ParamSpec]: + names = [p.name for p in old if not p.receiver and not p.keyword_only] + resolved: list[ParamSpec] = [] + seen: set[str] = set() + for spec in specs: + if spec.name in seen: + raise SignatureRefused( + cs.SIGNATURE_DUPLICATE_PARAM.format(name=spec.name) + ) + seen.add(spec.name) + index = spec.from_index + if spec.from_name is not None: + if spec.from_name not in names: + raise SignatureRefused( + cs.SIGNATURE_UNKNOWN_SOURCE.format( + source=spec.from_name, names=", ".join(names) + ) + ) + index = names.index(spec.from_name) + if index is not None and not 0 <= index < len(names): + raise SignatureRefused( + cs.SIGNATURE_UNKNOWN_SOURCE.format( + source=str(index), names=", ".join(names) + ) + ) + resolved.append(spec._replace(from_index=index, from_name=None)) + _refuse_duplicate_sources(resolved) + return resolved + + def _check_literals( + self, qn: str, specs: list[ParamSpec], old_names: list[str] + ) -> None: + """Refuse a default literal the declared parameter type cannot hold.""" + declared = self._declared_types(qn) + for spec in specs: + if spec.literal is None: + continue + annotation = spec.annotation + if ( + annotation is None + and spec.from_index is not None + and spec.from_index < len(declared) + ): + annotation = declared[spec.from_index] or None + if annotation is None: + continue + kind = _literal_kind(spec.literal) + if kind is not None and not _compatible(kind, annotation): + raise SignatureRefused( + cs.SIGNATURE_LITERAL_TYPE.format( + literal=spec.literal, name=spec.name, annotation=annotation + ) + ) + + def _declared_types(self, qn: str) -> list[str]: + rows = self.fetch_all( + cq.CYPHER_GRAPH_SIGNATURE, + {cs.KEY_PROJECT_PREFIX: f"{self.project}{cs.SEPARATOR_DOT}", cs.KEY_QN: qn}, + ) + for row in rows: + types = row.get(cs.KEY_PARAM_TYPES) + positional = row.get(cs.KEY_POSITIONAL_PARAMS) + if isinstance(types, list): + out = [str(t) for t in types] + # The receiver carries no annotation slot worth keeping. + if ( + isinstance(positional, list) + and positional + and positional[0] in _RECEIVERS + ): + out = out[1:] if len(out) == len(positional) else out + return out + return [] + + # --- planning --------------------------------------------------------------------- + + def plan( + self, qn: str, specs: Iterable[ParamSpec], allow_heuristic: bool = False + ) -> tuple[SignatureReport, Patcher]: + patcher = Patcher(self.repo_root) + hierarchy = _hierarchy(self.fetch_all, self.project, qn) + path, node, language, _source = self._definition(qn, patcher) + old = _parameters(node, language) + if any(p.node.type in _VARIADIC_PARAM_TYPES for p in old): + raise SignatureRefused(cs.SIGNATURE_VARIADIC.format(qn=qn)) + old_names = [p.name for p in old if not p.receiver and not p.keyword_only] + self._keyword_only = frozenset(p.name for p in old if p.keyword_only) + resolved = self._resolve_specs(specs, old) + self._check_literals(qn, resolved, old_names) + # Definitions across the hierarchy. Each member's own pre-edit + # parameter names are kept: an override may spell them differently + # from the member the request names, and its callers bind by ITS + # names, not the selected definition's. + member_names: dict[str, list[str]] = {} + member_keyword_only: dict[str, frozenset[str]] = {} + for member in hierarchy: + m_path, m_node, m_language, _ = ( + (path, node, language, _source) + if member == qn + else self._definition(member, patcher) + ) + m_old = _parameters(m_node, m_language) + member_names[member] = [ + p.name for p in m_old if not p.receiver and not p.keyword_only + ] + member_keyword_only[member] = frozenset( + p.name for p in m_old if p.keyword_only + ) + rewrite_definition(patcher, m_path, m_node, m_language, resolved) + # Every call site of every member. + sites: list[RewrittenSite] = [] + unmapped: list[UnmappedSite] = [] + for member in hierarchy: + # `Square().area(factor=2)` is spelled with Square's parameter + # name; matching it against Base's would read as an unknown + # keyword and leave the call unrewritten against a rewritten + # definition. + self._keyword_only = member_keyword_only.get(member, frozenset()) + for row in graph_query.callers(self.fetch_all, self.project, member): + rewrite_site( + patcher, + row, + member, + member_names.get(member, old_names), + resolved, + allow_heuristic, + self._parse, + self._keyword_only, + sites, + unmapped, + ) + self._keyword_only = frozenset(p.name for p in old if p.keyword_only) + report = SignatureReport( + qualified_name=qn, + hierarchy=tuple(hierarchy), + old_params=tuple(old_names), + new_params=tuple(spec.name for spec in resolved), + applied=False, + transaction_id="", + files=tuple(sorted(patcher.pending)), + sites=tuple(sorted(sites, key=lambda s: (s.path, s.line, s.col))), + unmapped=tuple(sorted(unmapped, key=lambda s: (s.path, s.line, s.col))), + diff="", + message=cs.SIGNATURE_PLANNED.format( + count=len(sites), unmapped=len(unmapped) + ), + ) + return report, patcher diff --git a/codebase_rag/editing/signature_rewrite.py b/codebase_rag/editing/signature_rewrite.py new file mode 100644 index 0000000000..6ca15b94aa --- /dev/null +++ b/codebase_rag/editing/signature_rewrite.py @@ -0,0 +1,139 @@ +"""Definition and call-site rewrite helpers for change_signature.""" + +from __future__ import annotations + +from collections.abc import Callable + +from tree_sitter import Node + +from .. import constants as cs +from .. import graph_query +from ..parsers.call_processor import _find_call_arguments_node +from ..types_defs import ResultRow +from .patcher import Patcher, PatcherError +from .signature_calls import _call_at, _map_arguments, _site_end +from .signature_params import ( + _check_default_order, + _parameters, + _render_param, + _rendered_old, + _restore_positional_only, + _text, +) +from .signature_types import ParamSpec, RewrittenSite, SignatureRefused, UnmappedSite + +_AMBIGUOUS = frozenset( + { + cs.EdgeResolution.HEURISTIC.value, + cs.EdgeResolution.OVERLOAD.value, + cs.EdgeResolution.DYNAMIC.value, + } +) + + +def rewrite_definition( + patcher: Patcher, + path: str, + node: Node, + language: cs.SupportedLanguage | None, + specs: list[ParamSpec], +) -> None: + params_node = node.child_by_field_name(cs.FIELD_PARAMETERS) + assert params_node is not None + old = _parameters(node, language) + positional = [p for p in old if not p.receiver and not p.keyword_only] + rendered = [_text(p.node) for p in old if p.receiver] + for spec in specs: + if spec.from_index is not None: + if spec.from_index >= len(positional): + # An override with fewer parameters than the target + # cannot supply the old value the spec maps from. + raise SignatureRefused( + cs.SIGNATURE_UNKNOWN_SOURCE.format( + source=spec.from_index, + names=", ".join(p.name for p in positional), + ) + ) + rendered.append(_rendered_old(positional[spec.from_index], spec, language)) + else: + rendered.append(_render_param(spec, language)) + if language == cs.SupportedLanguage.PYTHON: + _check_default_order(positional, specs) + _restore_positional_only(rendered, node, old, specs) + keyword_only = [p for p in old if p.keyword_only] + if keyword_only: + # The keyword-only section is not positional: it follows the + # rendered positionals behind its own `*`, exactly as written. + rendered.append("*") + rendered.extend(_text(p.node) for p in keyword_only) + text = "(" + ", ".join(rendered) + ")" + patcher.replace_span(path, (params_node.start_byte, params_node.end_byte), text) + + +def rewrite_site( + patcher: Patcher, + row: graph_query.CallSiteRow | ResultRow, + owner_qn: str, + old_names: list[str], + specs: list[ParamSpec], + allow_heuristic: bool, + parse_fn: Callable[[str, bytes], tuple[cs.SupportedLanguage | None, Node | None]], + keyword_only: frozenset[str], + sites: list[RewrittenSite], + unmapped: list[UnmappedSite], +) -> None: + path, line, col = row.get("path"), row.get("line"), row.get("col") + caller = str(row.get("qualified_name") or "") + resolution = row.get("resolution") + resolution_text = resolution if isinstance(resolution, str) else None + if ( + not isinstance(path, str) + or not isinstance(line, int) + or not isinstance(col, int) + ): + unmapped.append( + UnmappedSite( + path if isinstance(path, str) else "", + 0, + 0, + caller, + cs.SIGNATURE_UNLOCATABLE, + ) + ) + return + if resolution_text in _AMBIGUOUS and not allow_heuristic: + unmapped.append( + UnmappedSite( + path, + line, + col, + caller, + cs.SIGNATURE_GUESSED.format(resolution=resolution_text), + ) + ) + return + try: + source = patcher.source(path) + except PatcherError: + unmapped.append( + UnmappedSite(path, line, col, caller, cs.SIGNATURE_MISSING_FILE) + ) + return + language, root = parse_fn(path, source) + call = _call_at(root, line, col, _site_end(row)) if root is not None else None + args_node = _find_call_arguments_node(call) if call is not None else None + if args_node is None: + unmapped.append(UnmappedSite(path, line, col, caller, cs.SIGNATURE_NO_CALL)) + return + new_args = _map_arguments(args_node, old_names, specs, language, keyword_only) + if isinstance(new_args, str): + unmapped.append(UnmappedSite(path, line, col, caller, new_args)) + return + before = _text(args_node) + after = "(" + ", ".join(new_args) + ")" + if after != before: + patcher.replace_span(path, (args_node.start_byte, args_node.end_byte), after) + sites.append(RewrittenSite(path, line, col, caller, resolution_text, before, after)) + + +# --- applying --------------------------------------------------------------------- diff --git a/codebase_rag/editing/signature_types.py b/codebase_rag/editing/signature_types.py new file mode 100644 index 0000000000..9ec47f70c9 --- /dev/null +++ b/codebase_rag/editing/signature_types.py @@ -0,0 +1,103 @@ +"""Public data types and spec parsing for change_signature.""" + +from __future__ import annotations + +import re +from collections.abc import Iterable +from typing import NamedTuple + +from .. import constants as cs +from .contract import Verdict + +_SPEC_RE = re.compile( + r"^(?P[A-Za-z_][A-Za-z0-9_]*)" + r"(?::(?P[^=@]+))?" + r"(?:=(?P[^@]+))?" + r"(?:@(?P.+))?$" +) + + +class ParamSpec(NamedTuple): + """One parameter of the new signature and where its value comes from.""" + + name: str + from_index: int | None = None + from_name: str | None = None + literal: str | None = None + annotation: str | None = None + default: str | None = None + + @property + def unmapped(self) -> bool: + return ( + self.from_index is None and self.from_name is None and self.literal is None + ) + + +def parse_param_spec(text: str) -> ParamSpec: + """`name[:annotation][=default][@source]` as the CLI and MCP spell it. + + `@2` maps from old positional index 2, `@old` from the old parameter + `old`; a `=default` without a source is the literal every site that + passes nothing gains (and the definition's default); a bare name with + neither is unmapped. + """ + match = _SPEC_RE.match(text.strip()) + if match is None: + raise SignatureRefused(cs.SIGNATURE_BAD_SPEC.format(spec=text)) + name = match.group("name") + annotation = (match.group("annotation") or "").strip() or None + default = (match.group("default") or "").strip() or None + source = (match.group("source") or "").strip() or None + from_index = from_name = None + if source is not None: + if source.isdigit(): + from_index = int(source) + else: + from_name = source + literal = default if source is None and default is not None else None + return ParamSpec(name, from_index, from_name, literal, annotation, default) + + +class SignatureRefused(ValueError): + """The change cannot be planned as asked.""" + + +class UnmappedSite(NamedTuple): + path: str + line: int + col: int + owner: str + reason: str + + +class RewrittenSite(NamedTuple): + path: str + line: int + col: int + owner: str + resolution: str | None + before: str + after: str + + +class SignatureReport(NamedTuple): + qualified_name: str + hierarchy: tuple[str, ...] + old_params: tuple[str, ...] + new_params: tuple[str, ...] + applied: bool + transaction_id: str + files: tuple[str, ...] + sites: tuple[RewrittenSite, ...] + unmapped: tuple[UnmappedSite, ...] + diff: str + message: str + verdict: Verdict | None = None + + +# --- parameters of a definition ----------------------------------------------------- + + +def sites_for(sites: Iterable[NamedTuple]) -> list[dict[str, object]]: + return [dict(site._asdict()) for site in sites] diff --git a/evals/cgr_graph.py b/evals/cgr_graph.py index a5be63defd..bfb3e7c737 100644 --- a/evals/cgr_graph.py +++ b/evals/cgr_graph.py @@ -424,6 +424,19 @@ def _project_root_rows(self, params: PropertyDict) -> list[ResultRow]: def _graph_rows(self, query: str, params: PropertyDict) -> list[ResultRow]: qn = _str(params.get(cs.KEY_QN)) targets = self._graph_node_ids(qn) + if query == cq.CYPHER_GRAPH_SIGNATURE: + for label, uid in targets: + props = self.nodes[(label, uid)] + return [ + { + cs.KEY_POSITIONAL_PARAMS: _result( + props.get(cs.KEY_POSITIONAL_PARAMS) + ), + cs.KEY_PARAM_TYPES: _result(props.get(cs.KEY_PARAM_TYPES)), + cs.KEY_RETURN_TYPE: _result(props.get(cs.KEY_RETURN_TYPE)), + } + ] + return [] if query == cq.CYPHER_GRAPH_DEFINITION: for label, uid in targets: props = self.nodes[(label, uid)] @@ -574,6 +587,7 @@ def fetch_all( | cq.CYPHER_GRAPH_TYPE_EDGES | cq.CYPHER_GRAPH_OVERRIDES | cq.CYPHER_GRAPH_IMPORTERS + | cq.CYPHER_GRAPH_SIGNATURE ): return self._graph_rows(query, params or {}) case cs.CYPHER_ALL_FOLDER_PATHS: