diff --git a/codebase_rag/cli.py b/codebase_rag/cli.py index dc5be3c945..decce40800 100644 --- a/codebase_rag/cli.py +++ b/codebase_rag/cli.py @@ -1220,6 +1220,82 @@ def rename_command( raise typer.Exit(code=1) +@app.command( + name=ch.CLICommandName.CHANGE_SIGNATURE, + help=ch.CMD_CHANGE_SIGNATURE, + short_help=ch.CMD_CHANGE_SIGNATURE, + epilog=ch.EXAMPLES_CHANGE_SIGNATURE, + rich_help_panel=ch.PANEL_USE, +) +def change_signature_command( + qualified_name: str = typer.Argument(..., help=ch.HELP_CHANGE_SIGNATURE_QN), + new_params: list[str] = typer.Argument(..., help=ch.HELP_CHANGE_SIGNATURE_PARAMS), + repo_path: Path = typer.Option( + Path(cs.MCP_DEFAULT_DIRECTORY), + "--repo-path", + exists=True, + file_okay=False, + help=ch.HELP_GRAPH_REPO_PATH, + ), + project: str | None = typer.Option(None, "--project", help=ch.HELP_GRAPH_PROJECT), + mapping: list[str] | None = typer.Option( + None, "--map", help=ch.HELP_CHANGE_SIGNATURE_MAP + ), + allow_heuristic: bool = typer.Option( + False, "--allow-heuristic", help=ch.HELP_CHANGE_SIGNATURE_ALLOW_HEURISTIC + ), + dry_run: bool = typer.Option( + False, "--dry-run", help=ch.HELP_CHANGE_SIGNATURE_DRY_RUN + ), +) -> None: + from .editing.signature import ( + SignatureRefused, + change_signature, + parse_mapping, + sites_for, + unmapped_for, + ) + from .graph_cli import _project_and_fetch + + try: + parsed = parse_mapping(mapping or []) + except SignatureRefused as refused: + typer.echo(str(refused), err=True) + raise typer.Exit(code=1) from refused + name, fetch_all, ingestor = _project_and_fetch(project, repo_path) + with ingestor: # type: ignore[attr-defined] + parsers, queries = load_parsers() + updater = GraphUpdater( + ingestor=ingestor, # type: ignore[arg-type] + repo_path=repo_path.resolve(), + parsers=parsers, + queries=queries, + project_name=name, + ) + try: + report = change_signature( + repo_path.resolve(), + fetch_all, + name, + qualified_name, + new_params, + parsed, + allow_heuristic=allow_heuristic, + dry_run=dry_run, + reingest=updater.reingest, + ) + except SignatureRefused as refused: + typer.echo(str(refused), err=True) + raise typer.Exit(code=1) from refused + payload = dict(report._asdict()) + payload[cs.KEY_SITES] = sites_for(report.sites) + payload[cs.KEY_UNMAPPED] = unmapped_for(report.unmapped) + payload[cs.KEY_VERDICT] = report.verdict._asdict() if report.verdict else None + typer.echo(json.dumps(payload, indent=cs.MCP_JSON_INDENT, sort_keys=True)) + if not report.applied and not dry_run: + raise typer.Exit(code=1) + + @app.command( name=ch.CLICommandName.HELP, help=ch.CMD_HELP, diff --git a/codebase_rag/cli_help.py b/codebase_rag/cli_help.py index bca9bec57e..bd7750742e 100644 --- a/codebase_rag/cli_help.py +++ b/codebase_rag/cli_help.py @@ -21,6 +21,7 @@ class CLICommandName(StrEnum): GRAPH = "graph" CHECK = "check" RENAME = "rename" + CHANGE_SIGNATURE = "change-signature" STOP = "stop" STATUS = "status" HELP = "help" @@ -145,6 +146,31 @@ class CLICommandName(StrEnum): "Rewrite through heuristic, overload and trace-only sites as well." ) HELP_RENAME_DRY_RUN = "Print the plan and diff without writing anything." +CMD_CHANGE_SIGNATURE = ( + "Change a Python definition's parameter list and rewrite every call site " + "the graph knows; sites the mapping cannot complete are listed, not touched." +) +EXAMPLES_CHANGE_SIGNATURE = ( + "Examples:\n" + " cgr change-signature myproj.pkg.util.helper a 'n: int' b --map n==1 --dry-run\n" + " cgr change-signature myproj.pkg.util.helper b a\n" + " cgr change-signature myproj.pkg.util.helper 'times: int' 'text: str' " + "--map times=0 --map text=1" +) +HELP_CHANGE_SIGNATURE_QN = "Qualified name of the function or method to change." +HELP_CHANGE_SIGNATURE_PARAMS = ( + "The new parameters in order: NAME, 'NAME: TYPE', 'NAME = DEFAULT' or " + "'NAME: TYPE = DEFAULT'; a bare old name carries it over unchanged." +) +HELP_CHANGE_SIGNATURE_MAP = ( + "NEW=SOURCE, repeatable: SOURCE is an old parameter name, its 0-based " + "index, or =LITERAL to pass that text at every site (so NEW==1 inserts 1)." +) +HELP_CHANGE_SIGNATURE_ALLOW_HEURISTIC = ( + "Rewrite sites the graph resolved heuristically, by overload fan-out or by " + "a trace as well, instead of listing them as unmapped." +) +HELP_CHANGE_SIGNATURE_DRY_RUN = "Print the plan and diff without writing anything." CMD_TRACE_INGEST = "Resolve a trace file against a project and write dynamic edges" CMD_TRACE_CONVERT = "Convert a V8 .cpuprofile (node --cpu-prof) to a trace file" @@ -466,6 +492,7 @@ class CLICommandName(StrEnum): CLICommandName.GRAPH: CMD_GRAPH, CLICommandName.CHECK: CMD_CHECK, CLICommandName.RENAME: CMD_RENAME, + CLICommandName.CHANGE_SIGNATURE: CMD_CHANGE_SIGNATURE, CLICommandName.WORKSPACE: CMD_WORKSPACE, CLICommandName.STOP: CMD_STOP, CLICommandName.STATUS: CMD_STATUS, diff --git a/codebase_rag/constants/ast_python.py b/codebase_rag/constants/ast_python.py index 7c6a1e6c4c..49975ae7b4 100644 --- a/codebase_rag/constants/ast_python.py +++ b/codebase_rag/constants/ast_python.py @@ -51,6 +51,24 @@ TS_PY_DICTIONARY_SPLAT_PATTERN = "dictionary_splat_pattern" TS_PY_POSITIONAL_SEPARATOR = "positional_separator" TS_PY_KEYWORD_SEPARATOR = "keyword_separator" +# A definition's parameter list node, and the splats an argument list can +# carry (`f(*args)`, `f(**opts)`); the `_pattern` forms above are the +# definition-side spellings. +TS_PY_PARAMETERS = "parameters" +TS_PY_LIST_SPLAT = "list_splat" +TS_PY_DICTIONARY_SPLAT = "dictionary_splat" +# Scopes nested in a function body, where a parameter's name may be re-bound +# rather than read, and the statements that re-bind a name in place. +TS_PY_SET_COMPREHENSION = "set_comprehension" +TS_PY_DICTIONARY_COMPREHENSION = "dictionary_comprehension" +TS_PY_GENERATOR_EXPRESSION = "generator_expression" +TS_PY_NONLOCAL_STATEMENT = "nonlocal_statement" +# `typing` names a literal-versus-annotation check understands. +PY_TYPING_OPTIONAL = "Optional" +PY_TYPING_UNION = "Union" +# The first parameter of a method that names the instance or class; a +# signature change keeps it in place and never remaps it. +PY_RECEIVER_NAMES = frozenset({"self", "cls"}) TS_PY_SUBSCRIPT = "subscript" # The `subscript` node's index field (`os.environ["K"]` -> the `"K"` string). TS_PY_FIELD_SUBSCRIPT = "subscript" diff --git a/codebase_rag/constants/cli.py b/codebase_rag/constants/cli.py index 185e4669ed..9791b44e0e 100644 --- a/codebase_rag/constants/cli.py +++ b/codebase_rag/constants/cli.py @@ -605,6 +605,109 @@ class DiffMarker: RENAME_CONTRACT_UNMEASURED = ( "Rename applied, but its postcondition could not be measured: {error}" ) +# Change signature (issue #1533). +SIGNATURE_PARAM_PROBE = "def _({text}): pass" +SIGNATURE_NOT_PYTHON = ( + "Cannot change the signature of {qn}: {path} is not Python, and only Python " + "definitions are supported for now (issue #{issue})" +) +SIGNATURE_DEFINITION_UNREADABLE = ( + "Cannot change the signature of {qn}: its file {path} cannot be read " + "({error}). The index names a file the tree no longer has; re-index and retry." +) +SIGNATURE_NO_HEADER = "Could not locate the definition of {qn} in {path}" +SIGNATURE_UNSUPPORTED_PARAMS = ( + "Cannot change the signature of {qn}: `{text}` is not a plain " + "positional-or-keyword parameter, and only those can be remapped" +) +SIGNATURE_UNUSUAL_RECEIVER = ( + "Cannot change the signature of {qn}: its first parameter `{name}` is not " + "self or cls, so the receiver cannot be told from the parameters; rename " + "it to self or cls first" +) +SIGNATURE_STAGE_FAILED = "Cannot stage the signature change: {error}" +SIGNATURE_BAD_PARAM = ( + "Not a parameter: `{text}` (expected `name`, `name: type`, `name = default` " + "or `name: type = default`)" +) +SIGNATURE_DUPLICATE_NEW = "Parameter {name} is listed twice" +SIGNATURE_REQUIRED_AFTER_DEFAULT = ( + "Parameter {name} has no default but comes after a defaulted parameter" +) +SIGNATURE_BODY_REBINDS = ( + "Cannot rename parameter {old} of {qn} to {new}: `{old}` is used inside a " + "nested scope, or re-bound by a global, nonlocal or import statement, in " + "its body; rename it by hand first" +) +SIGNATURE_DEFAULT_REFERENCES_RENAMED = ( + "Cannot rename parameter {old} of {qn} to {new}: the default of `{param}` " + "reads `{old}`, which is evaluated at definition time and would raise " + "NameError; rewrite that default by hand first" +) +SIGNATURE_DROPPED_STILL_READ = ( + "Cannot drop parameter {name} of {qn}: its body still reads `{name}`, " + "which would raise NameError when called; remove that use by hand first" +) +SIGNATURE_BODY_NAME_TAKEN = ( + "Cannot rename parameter {old} of {qn} to {new}: `{new}` is already used " + "in its body" +) +SIGNATURE_HIERARCHY_MISMATCH = ( + "Cannot change the signature of {qn}: its override {member} declares " + "({theirs}) where {qn} declares ({ours}); make the hierarchy agree first" +) +SIGNATURE_MAPPING_UNKNOWN_NEW = ( + "The mapping names {name}, which is not a new parameter ({names})" +) +SIGNATURE_MAPPING_UNKNOWN_OLD = ( + "The mapping feeds {name} from `{source}`, which is neither an old parameter " + "({names}), an index into them, nor a `=literal`" +) +SIGNATURE_MAPPING_BAD_INDEX = ( + "The mapping feeds {name} from index {index}, but only {count} old " + "parameter(s) exist" +) +SIGNATURE_MAPPING_EMPTY_LITERAL = "The mapping gives {name} an empty literal" +SIGNATURE_MAPPING_DUPLICATE = ( + "Old parameter {old} cannot feed both {first} and {second}" +) +SIGNATURE_LITERAL_MISMATCH = ( + "{literal} does not fit the declared type of {name} ({annotation})" +) +SIGNATURE_BAD_MAP = "Mapping entries take the form NEW=SOURCE, got `{entry}`" +SIGNATURE_SITE_NO_LOCATION = "site carries no location ({resolution})" +SIGNATURE_SITE_GUESSED = ( + "site was resolved by {resolution}; pass allow_heuristic to rewrite it" +) +SIGNATURE_SITE_NOT_PYTHON = "{path} is not Python" +SIGNATURE_SITE_UNREADABLE_FILE = "file cannot be read ({error})" +SIGNATURE_SITE_NO_CALL = "no call found at the recorded position" +SIGNATURE_SITE_UNREADABLE = "arguments cannot be read positionally: `{text}`" +SIGNATURE_SITE_UNKNOWN_KEYWORD = "passes keyword `{name}`, which is not declared" +SIGNATURE_SITE_DUPLICATE = "passes `{name}` more than once" +SIGNATURE_SITE_TOO_MANY = ( + "passes {given} positional arguments but {declared} are declared" +) +SIGNATURE_SITE_NO_VALUE = "no value for `{name}`; map it or give it a default" +SIGNATURE_SITE_OVERLAPS = "its arguments overlap another edit of this change" +SIGNATURE_PLANNED = "{count} site(s) would be rewritten, {skipped} 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_ROLLBACK_REFUSED = ( + "Signature change kept: its postcondition failed ({reasons}) but a later " + "edit was recorded on top of it, so it was not rolled back; undo the later " + "edit and rerun" +) +SIGNATURE_ROLLBACK_UNMEASURED = ( + "Signature change rolled back after its postcondition failed ({reasons}), " + "but the graph could not be re-ingested afterwards ({error}); rebuild the " + "graph before the next graph-backed operation" +) +SIGNATURE_CONTRACT_UNMEASURED = ( + "Signature change applied, but its postcondition could not be measured: {error}" +) # Postcondition contract (issue #1531). CONTRACT_OP_RENAME = "rename" CONTRACT_OP_CHANGE_SIGNATURE = "change_signature" diff --git a/codebase_rag/constants/core.py b/codebase_rag/constants/core.py index 65f10a2c4f..ef4e74305b 100644 --- a/codebase_rag/constants/core.py +++ b/codebase_rag/constants/core.py @@ -102,6 +102,7 @@ class EventType(StrEnum): CHAR_SPACE = " " SEPARATOR_COMMA_SPACE = ", " +SEPARATOR_SEMICOLON_SPACE = "; " PUNCTUATION_TYPES = (CHAR_PAREN_OPEN, CHAR_PAREN_CLOSE, CHAR_COMMA) REGEX_METHOD_CHAIN_SUFFIX = r"\)\.[^)]*$" diff --git a/codebase_rag/constants/graph.py b/codebase_rag/constants/graph.py index e848ee44dc..59e672c482 100644 --- a/codebase_rag/constants/graph.py +++ b/codebase_rag/constants/graph.py @@ -59,6 +59,10 @@ # Rename report fields (issue #1532). KEY_SITES = "sites" KEY_AMBIGUOUS = "ambiguous" +# Change-signature report fields (issue #1533); `verdict` is shared +# with rename. +KEY_UNMAPPED = "unmapped" +KEY_VERDICT = "verdict" KEY_STRUCTURAL_DELTA = "structural_delta" KEY_DISPATCH_LITERAL = "dispatch_literal" diff --git a/codebase_rag/constants/mcp.py b/codebase_rag/constants/mcp.py index 036b955fc4..4a9e9a0405 100644 --- a/codebase_rag/constants/mcp.py +++ b/codebase_rag/constants/mcp.py @@ -19,8 +19,9 @@ class MCPToolName(StrEnum): OVERRIDES = "overrides" IMPORTERS = "importers" TESTS_REACHING = "tests_reaching" - # Graph-driven edit operations (issue #1532). + # Graph-driven edit operations (issues #1532, #1533). RENAME = "rename" + CHANGE_SIGNATURE = "change_signature" QUERY_CODE_GRAPH = "query_code_graph" GET_CODE_SNIPPET = "get_code_snippet" SURGICAL_REPLACE_CODE = "surgical_replace_code" @@ -110,6 +111,8 @@ class MCPParamName(StrEnum): MODULE_QN = "module_qualified_name" NEW_NAME = "new_name" ALLOW_HEURISTIC = "allow_heuristic" + NEW_PARAMS = "new_params" + MAPPING = "mapping" # MCP server constants diff --git a/codebase_rag/editing/__init__.py b/codebase_rag/editing/__init__.py index 3786e8b039..66bd2f8214 100644 --- a/codebase_rag/editing/__init__.py +++ b/codebase_rag/editing/__init__.py @@ -1,5 +1,7 @@ """Graph-aware editing primitives: span-preserving patchers (issue #1529), -transactions (issue #1528) and import rewriting (issue #1530).""" +transactions (issue #1528), import rewriting (issue #1530) and the graph +operations built on them: rename (issue #1532) and change_signature (issue +#1533).""" from .contract import ( Expectation, diff --git a/codebase_rag/editing/contract.py b/codebase_rag/editing/contract.py index 0e3b974499..495a099df3 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, ) @@ -111,6 +114,15 @@ def _site_key(site: Mapping[str, object]) -> str: return f"{site.get(cs.KEY_PATH)}:{site.get(cs.KEY_LINE)}" +def _call_key(site: Mapping[str, object]) -> str: + """A call's identity for the signature checks. + + Two calls can share a line (`helper(1) + helper(2)`), and rewriting or + listing one must not cover the other, so the column is part of the key. + """ + return f"{_site_key(site)}:{site.get(cs.KEY_COL)}" + + def _carried_by_ancestor( pair: tuple[str, str], expected: Iterable[tuple[str, str]] ) -> bool: @@ -200,7 +212,7 @@ def _changed_site_fault( loops plus these three verdicts put that function one point over the threshold. The decision is per site and reads better named anyway. """ - key = _site_key(site) + key = _call_key(site) verdict = site["verdict"] if verdict == cs.DELTA_ARITY_OK or key in unmapped: return None @@ -232,7 +244,7 @@ def _check_sites_mapped( if fault := _changed_site_fault(site, unmapped, rewritten): bad.append(fault) for site in delta["arity_findings"]: - key = _site_key(site) + key = _call_key(site) if key not in unmapped: bad.append(f"{key} ({site['verdict']})") if bad: @@ -303,7 +315,9 @@ def verify( """Pass or fail the delta against the expectation, with reasons. `rewritten` lists the sites the operation rewrote as `(path:line, - resolution)`; `parse_failures` the files the transaction found not to + resolution)`, or `(path:line:col, resolution)` for a signature change, + whose per-site checks key on the column; `parse_failures` the files the + transaction found not to parse (the transaction refuses those itself, so a caller normally passes none, but the contract states the rule in one place). """ diff --git a/codebase_rag/editing/rename.py b/codebase_rag/editing/rename.py index dcca71cfaf..a661adc07e 100644 --- a/codebase_rag/editing/rename.py +++ b/codebase_rag/editing/rename.py @@ -44,6 +44,7 @@ from .contract import Reingest, Verdict, measure, rename_expectation, verify from .imports import ANY_MODULE, ImportRewriter, ImportSite, SymbolMove, _imported from .patcher import Patcher, PatcherError, line_col_to_byte +from .sites import AMBIGUOUS, call_node_at, hierarchy from .transaction import ( EditTransaction, StagedTree, @@ -55,13 +56,6 @@ QueryFn = Callable[[str, PropertyDict | None], list[ResultRow]] -_AMBIGUOUS = frozenset( - { - cs.EdgeResolution.HEURISTIC.value, - cs.EdgeResolution.OVERLOAD.value, - cs.EdgeResolution.DYNAMIC.value, - } -) _IDENTIFIER_RE = r"(? list[str]: - """`qn` plus every method it overrides or is overridden by, transitively.""" - seen: list[str] = [qn] - frontier = [qn] - while frontier: - current = frontier.pop() - for row in graph_query.overrides(fetch_all, project, current): - other = row["qualified_name"] - if other not in seen: - seen.append(other) - frontier.append(other) - return seen - - def _name_token( source: bytes, language: cs.SupportedLanguage | None, @@ -192,39 +172,6 @@ def _name_token( return None -def _best_call_at( - root: Node, line: int, col: int, recorded_end: tuple[int, int] | None -) -> Node | None: - """The call node at (line, col) that the graph site refers to. - - Several calls can share a start point -- `helper(helper(1))`, - `helper(2).upper()`, and both links of `obj.helper(1).helper(2)` -- so the - right one is the call ending where the site recorded its end, or the - outermost when no end was recorded. - - Extracted from `_callee_span` to keep it under the cognitive complexity - limit (S3776). This walk is where that complexity lives: a loop with - three levels of nested branching, and nesting multiplies the cost. - Extracting the straight-line setup around it would not have helped. - """ - best: Node | None = None - stack: list[Node] = [root] - while stack: - node = stack.pop() - if node.start_point == (line - 1, col): - func = node.child_by_field_name(cs.FIELD_FUNCTION) - if func is not None: - if node.end_point == recorded_end: - return node - if best is None or ( - best.end_point != recorded_end and node.end_byte > best.end_byte - ): - best = node - if node.start_point[0] <= line - 1 <= node.end_point[0]: - stack.extend(node.children) - return best - - def _callee_span( source: bytes, language: cs.SupportedLanguage | None, @@ -252,7 +199,7 @@ def _callee_span( if end_line is not None and end_col is not None else None ) - best = _best_call_at(root, line, col, recorded_end) + best = call_node_at(root, line, col, recorded_end) if best is None: return None func = best.child_by_field_name(cs.FIELD_FUNCTION) @@ -609,11 +556,11 @@ def plan( """Collect everything a rename touches; refuse on ambiguity.""" if not re.fullmatch(r"[A-Za-z_]\w*", new_name): raise RenameRefused(cs.RENAME_BAD_NAME.format(name=new_name), [], []) - hierarchy = _hierarchy(self.fetch_all, self.project, qn) + members = hierarchy(self.fetch_all, self.project, qn) sites: list[RenameSite] = [] unlocatable: list[str] = [] old_name: str | None = None - for member in hierarchy: + for member in members: member_sites, member_unlocatable, member_name, _label = self._collect( member ) @@ -645,7 +592,7 @@ def plan( unlocatable, ) ambiguous = [ - s for s in sites if s.resolution in _AMBIGUOUS or s.resolution == _CHAIN + s for s in sites if s.resolution in AMBIGUOUS or s.resolution == _CHAIN ] if ambiguous and not allow_heuristic: raise RenameRefused( @@ -653,7 +600,7 @@ def plan( ambiguous, unlocatable, ) - for member in hierarchy: + for member in members: for site, module in self._import_sites(member, old_name): sites.append( RenameSite( @@ -680,7 +627,7 @@ def plan( ambiguous=tuple(ambiguous), unlocatable=tuple(unlocatable), doc_mentions=tuple(self._doc_mentions(old_name)), - hierarchy=tuple(hierarchy), + hierarchy=tuple(members), diff="", message=cs.RENAME_PLANNED.format(count=len(sites)), ) diff --git a/codebase_rag/editing/signature.py b/codebase_rag/editing/signature.py new file mode 100644 index 0000000000..5e325ff9d4 --- /dev/null +++ b/codebase_rag/editing/signature.py @@ -0,0 +1,1322 @@ +"""Edit algebra op 2: `change_signature(qn, new_params, mapping)` (issue #1533). + +A parameter list changed by hand is where call sites drift: one caller +missed, a positional value silently shifted into the wrong slot. The graph +knows every site, so the change is a graph operation: + +1. read the definition's parameter list and, for a method on a hierarchy, + the list of every override in both directions (they must agree); +2. work out where each new parameter's value comes from at a call site: an + old parameter carried by name or by index, or a literal the mapping + supplies for every site; +3. rewrite the definition(s) and every graph-known call site through the + span patcher (issue #1529), stage the result in a transaction (issue + #1528), and commit, or roll back and report why. + +A site the mapping cannot complete -- a value the caller never passed for a +parameter without a default, a splat, a surplus argument, a keyword the +definition does not declare -- is left exactly as written and reported as +`unmapped`; so is a site the graph bound by guesswork (`heuristic`, +`overload`, `dynamic`) unless the caller accepts the risk with +`allow_heuristic`. With a re-ingest the applied change is measured through +the structural delta and undone when its postcondition contract (issue +#1531) fails: every site of the changed signature must read as mapped or +be in the unmapped list. + +Python only for now: the header and the argument lists are read with the +Python grammar, and a definition in any other language refuses (issue +#1908). +""" + +from __future__ import annotations + +import ast +import re +from collections.abc import Iterable, Mapping, Sequence +from pathlib import Path +from typing import NamedTuple + +from loguru import logger +from tree_sitter import Node, Parser + +from .. import constants as cs +from .. import graph_query +from ..graph_query import QueryFn +from ..graph_updater import ReingestAborted +from ..language_spec import get_language_for_extension +from ..parser_loader import load_parsers +from ..parsers.utils import _is_static_decorator +from .contract import Reingest, Verdict, change_signature_expectation, measure, verify +from .patcher import Patcher, PatcherError +from .sites import AMBIGUOUS, call_node_at, hierarchy +from .transaction import EditTransaction, TransactionConflict, undo_transaction + +# Issue tracking support for definitions in other languages. +LANGUAGES_ISSUE = 1908 + +_IDENTIFIER_RE = re.compile(r"[A-Za-z_]\w*") +_LITERAL_PREFIX = "=" +_DEFINITION = "definition" +_CALL = "call" +# Runtime types a literal may have under a builtin annotation. `float` +# admits an int the way the type checkers do; `bool` is not an `int` here +# because `flag: int = True` is almost always a mistake. +_BUILTIN_TYPES: dict[str, frozenset[type]] = { + int.__name__: frozenset({int}), + float.__name__: frozenset({int, float}), + complex.__name__: frozenset({int, float, complex}), + str.__name__: frozenset({str}), + bool.__name__: frozenset({bool}), + bytes.__name__: frozenset({bytes}), + list.__name__: frozenset({list}), + tuple.__name__: frozenset({tuple}), + dict.__name__: frozenset({dict}), + set.__name__: frozenset({set}), + frozenset.__name__: frozenset({frozenset}), +} + + +class ParamSpec(NamedTuple): + name: str + text: str # as written in a header: `b: str = 'x'` + annotation: str | None + has_default: bool + + +class SignatureSite(NamedTuple): + kind: str # "definition" or "call" + path: str + line: int + col: int + owner: str # the qualified name the site belongs to + resolution: str | None + + +class UnmappedSite(NamedTuple): + owner: str + path: str + line: int | None + col: int | None + reason: str + + +class SignatureRefused(ValueError): + """The change cannot be stated safely; nothing was changed.""" + + +class SignatureReport(NamedTuple): + qualified_name: str + old_params: tuple[str, ...] + new_params: tuple[str, ...] + applied: bool + transaction_id: str + files: tuple[str, ...] + sites: tuple[SignatureSite, ...] + unmapped: tuple[UnmappedSite, ...] + hierarchy: tuple[str, ...] + diff: str + message: str + verdict: Verdict | None = None + # True when a rollback re-ingest failed after the files were restored: + # the graph may hold a partial picture and must be rebuilt. + graph_incomplete: bool = False + + +class _Source(NamedTuple): + """Where a new parameter's value comes from at a call site.""" + + index: int | None = None # an old parameter, by position + literal: str | None = None # the same text at every site + + +class _Header(NamedTuple): + qn: str + path: str + span: tuple[int, int] # byte span of the `parameters` node + line: int + col: int + receiver: str | None # `self`/`cls`, kept verbatim and never remapped + params: list[ParamSpec] + function: Node # the `function_definition`, for its body + source: bytes + + +class _Binding(NamedTuple): + text: str + keyword: bool + position: int # index among the site's arguments, for keyword order + span: tuple[int, int] # byte span of the value, for edits nested in it + + +class _Edit(NamedTuple): + path: str + span: tuple[int, int] + text: str + + +class _Candidate(NamedTuple): + """A call site the graph located and the mapping can read, not yet rendered.""" + + site: SignatureSite + span: tuple[int, int] # byte span of the argument list + text: str # the argument list as written + bound: dict[int, _Binding] + + +class _Unmapped(Exception): + """A call site the mapping cannot rewrite; carries the reason.""" + + +# --- the mapping ------------------------------------------------------------------ + + +def parse_mapping(entries: Iterable[str]) -> dict[str, str]: + """`NEW=SOURCE` entries as the CLI takes them, into the mapping. + + The split is on the first `=`, so `n==1` maps `n` to the literal `1`. + """ + mapping: dict[str, str] = {} + for entry in entries: + new, sep, source = entry.partition(_LITERAL_PREFIX) + if not sep or not new: + raise SignatureRefused(cs.SIGNATURE_BAD_MAP.format(entry=entry)) + mapping[new] = source + return mapping + + +def _resolve_sources( + new: Sequence[ParamSpec], old: Sequence[ParamSpec], mapping: Mapping[str, str] +) -> list[_Source | None]: + """One source per new parameter; None where the mapping says nothing.""" + old_names = [p.name for p in old] + new_names = [p.name for p in new] + for name in mapping: + if name not in new_names: + raise SignatureRefused( + cs.SIGNATURE_MAPPING_UNKNOWN_NEW.format( + name=name, names=cs.SEPARATOR_COMMA_SPACE.join(new_names) + ) + ) + sources: list[_Source | None] = [] + fed: dict[int, str] = {} + for spec in new: + source = _source_for(spec, mapping.get(spec.name), old_names) + if source is not None and source.index is not None: + if source.index in fed: + raise SignatureRefused( + cs.SIGNATURE_MAPPING_DUPLICATE.format( + old=old_names[source.index], + first=fed[source.index], + second=spec.name, + ) + ) + fed[source.index] = spec.name + sources.append(source) + return sources + + +def _source_for( + spec: ParamSpec, text: str | None, old_names: Sequence[str] +) -> _Source | None: + if text is None: + # Not mentioned: the old parameter of the same name, if any. + if spec.name in old_names: + return _Source(index=old_names.index(spec.name)) + return None + if text.startswith(_LITERAL_PREFIX): + literal = text[len(_LITERAL_PREFIX) :].strip() + if not literal: + raise SignatureRefused( + cs.SIGNATURE_MAPPING_EMPTY_LITERAL.format(name=spec.name) + ) + return _Source(literal=literal) + if text.isdigit(): + index = int(text) + if index >= len(old_names): + raise SignatureRefused( + cs.SIGNATURE_MAPPING_BAD_INDEX.format( + name=spec.name, index=index, count=len(old_names) + ) + ) + return _Source(index=index) + if text in old_names: + return _Source(index=old_names.index(text)) + raise SignatureRefused( + cs.SIGNATURE_MAPPING_UNKNOWN_OLD.format( + name=spec.name, + source=text, + names=cs.SEPARATOR_COMMA_SPACE.join(old_names), + ) + ) + + +# --- the new parameter list ----------------------------------------------------- + + +def _parse_new_param(text: str, old: Mapping[str, ParamSpec]) -> ParamSpec: + """A new parameter as the caller spelled it. + + A bare identifier naming an old parameter carries that parameter over + with its annotation and default; anything else must parse as exactly + one plain positional-or-keyword parameter. + """ + text = text.strip() + if _IDENTIFIER_RE.fullmatch(text) and text in old: + return old[text] + probe = cs.SIGNATURE_PARAM_PROBE.format(text=text) + try: + module = ast.parse(probe) + except (SyntaxError, ValueError) as error: + raise SignatureRefused(cs.SIGNATURE_BAD_PARAM.format(text=text)) from error + function = module.body[0] if len(module.body) == 1 else None + if not isinstance(function, ast.FunctionDef): + raise SignatureRefused(cs.SIGNATURE_BAD_PARAM.format(text=text)) + args = function.args + plain = ( + len(args.args) == 1 + and not args.posonlyargs + and not args.kwonlyargs + and args.vararg is None + and args.kwarg is None + ) + if not plain: + raise SignatureRefused(cs.SIGNATURE_BAD_PARAM.format(text=text)) + arg = args.args[0] + annotation = ( + ast.get_source_segment(probe, arg.annotation) + if arg.annotation is not None + else None + ) + return ParamSpec(arg.arg, text, annotation, bool(args.defaults)) + + +def _accepted_types(node: ast.expr) -> set[type] | None: + """Runtime types a literal may take under this annotation. + + None when the annotation is not one this check reads (a project class, a + `typing` alias other than Optional/Union, an attribute): those are not + refused, merely not checked. + """ + if isinstance(node, ast.Constant) and node.value is None: + return {type(None)} + if isinstance(node, ast.Name): + accepted = _BUILTIN_TYPES.get(node.id) + return set(accepted) if accepted is not None else None + if isinstance(node, ast.BinOp) and isinstance(node.op, ast.BitOr): + return _union_of([node.left, node.right]) + if isinstance(node, ast.Subscript) and isinstance(node.value, ast.Name): + return _subscript_types(node.value, node.slice) + return None + + +def _subscript_types(head: ast.Name, index: ast.expr) -> set[type] | None: + """`Optional[X]`, `Union[X, Y]`, or a subscripted builtin like `list[int]`.""" + if head.id == cs.PY_TYPING_OPTIONAL: + inner = _accepted_types(index) + return inner | {type(None)} if inner is not None else None + if head.id == cs.PY_TYPING_UNION: + members = list(index.elts) if isinstance(index, ast.Tuple) else [index] + return _union_of(members) + # `list[int]` is a list; the element type is not checked. + return _accepted_types(head) + + +def _union_of(members: Iterable[ast.expr]) -> set[type] | None: + accepted: set[type] = set() + for member in members: + part = _accepted_types(member) + if part is None: + return None + accepted |= part + return accepted + + +def _check_literal(spec: ParamSpec, literal: str) -> None: + """Refuse a literal that cannot be a value of the declared type.""" + if spec.annotation is None: + return + try: + value = ast.literal_eval(literal) + annotation = ast.parse(spec.annotation, mode="eval").body + except (ValueError, SyntaxError, TypeError, MemoryError, RecursionError): + # Not a literal (`=LIMIT`), or an annotation Python cannot parse: + # nothing to compare. + return + accepted = _accepted_types(annotation) + if accepted is not None and type(value) not in accepted: + raise SignatureRefused( + cs.SIGNATURE_LITERAL_MISMATCH.format( + literal=literal, name=spec.name, annotation=spec.annotation + ) + ) + + +def _new_specs(new_params: Sequence[str], old: Sequence[ParamSpec]) -> list[ParamSpec]: + by_name = {spec.name: spec for spec in old} + specs = [_parse_new_param(text, by_name) for text in new_params] + seen: set[str] = set() + defaulted = False + for spec in specs: + if spec.name in seen: + raise SignatureRefused(cs.SIGNATURE_DUPLICATE_NEW.format(name=spec.name)) + seen.add(spec.name) + if spec.has_default: + defaulted = True + elif defaulted: + raise SignatureRefused( + cs.SIGNATURE_REQUIRED_AFTER_DEFAULT.format(name=spec.name) + ) + return specs + + +# --- the header ----------------------------------------------------------------- + + +def _text(node: Node, source: bytes) -> str: + return source[node.start_byte : node.end_byte].decode(cs.ENCODING_UTF8) + + +def _is_static(function: Node, source: bytes) -> bool: + """Decorated `@staticmethod`: every parameter is one the caller passes. + + Read the way the indexer reads decorators, by the last name, so the + qualified `@builtins.staticmethod` counts too. + """ + parent = function.parent + if parent is None or parent.type != cs.TS_PY_DECORATED_DEFINITION: + return False + return _is_static_decorator( + [ + _text(child, source) + for child in parent.children + if child.type == cs.TS_PY_DECORATOR + ] + ) + + +def _find_definition(root: Node, name: str, start: int, end: int) -> Node | None: + """The `function_definition` named `name` starting within lines start..end.""" + wanted = name.encode(cs.ENCODING_UTF8) + stack: list[Node] = [root] + while stack: + node = stack.pop() + if ( + node.type == cs.TS_PY_FUNCTION_DEFINITION + and start <= node.start_point[0] + 1 <= end + ): + ident = node.child_by_field_name(cs.FIELD_NAME) + if ident is not None and ident.text == wanted: + return node + if node.start_point[0] < end and node.end_point[0] >= start - 1: + stack.extend(reversed(node.children)) + return None + + +def _param_spec(node: Node, source: bytes) -> ParamSpec | None: + """A plain positional-or-keyword parameter, or None for anything else.""" + text = _text(node, source) + if node.type == cs.TS_PY_IDENTIFIER: + return ParamSpec(text, text, None, False) + if node.type == cs.TS_PY_TYPED_PARAMETER: + # `*args: int` is a typed_parameter over a splat pattern. + first = node.named_children[0] if node.named_children else None + annotation = node.child_by_field_name(cs.FIELD_TYPE) + if first is None or first.type != cs.TS_PY_IDENTIFIER or annotation is None: + return None + return ParamSpec(_text(first, source), text, _text(annotation, source), False) + if node.type in (cs.TS_PY_DEFAULT_PARAMETER, cs.TS_PY_TYPED_DEFAULT_PARAMETER): + name = node.child_by_field_name(cs.FIELD_NAME) + if name is None or name.type != cs.TS_PY_IDENTIFIER: + return None + annotation = node.child_by_field_name(cs.FIELD_TYPE) + return ParamSpec( + _text(name, source), + text, + _text(annotation, source) if annotation is not None else None, + True, + ) + return None + + +# --- renamed parameters in the body ---------------------------------------------- + +# Scopes of their own inside a function body. A comprehension binds only its +# `for` targets, so one that merely reads the parameter follows the rename; +# a nested function, lambda or class can bind the name in ways the body walk +# cannot see (an assignment there is a fresh local), so any use of the name +# inside one refuses. +_COMPREHENSIONS = frozenset( + { + cs.TS_PY_LIST_COMPREHENSION, + cs.TS_PY_SET_COMPREHENSION, + cs.TS_PY_DICTIONARY_COMPREHENSION, + cs.TS_PY_GENERATOR_EXPRESSION, + } +) +_OWN_SCOPES = frozenset( + {cs.TS_PY_FUNCTION_DEFINITION, cs.TS_PY_LAMBDA, cs.TS_PY_CLASS_DEFINITION} +) +_REBINDING_STATEMENTS = frozenset( + { + cs.TS_PY_GLOBAL_STATEMENT, + cs.TS_PY_NONLOCAL_STATEMENT, + cs.TS_PY_IMPORT_STATEMENT, + cs.TS_PY_IMPORT_FROM_STATEMENT, + } +) +_OPAQUE_TO_THE_WALK = _OWN_SCOPES | _REBINDING_STATEMENTS + + +def _is_a_reference(node: Node) -> bool: + """A bare identifier, not the `.attr` of an attribute or a keyword's name.""" + parent = node.parent + if parent is None: + return True + if parent.type == cs.TS_PY_ATTRIBUTE: + return parent.child_by_field_name(cs.TS_PY_FIELD_ATTRIBUTE) != node + if parent.type == cs.TS_PY_KEYWORD_ARGUMENT: + return parent.child_by_field_name(cs.FIELD_NAME) != node + return True + + +def _mentions(node: Node, name: bytes) -> bool: + stack = [node] + while stack: + current = stack.pop() + if current.type == cs.TS_PY_IDENTIFIER and current.text == name: + return True + stack.extend(current.children) + return False + + +def _comprehension_binds(node: Node, name: bytes) -> bool: + return any( + child.type == cs.TS_PY_FOR_IN_CLAUSE + and (left := child.child_by_field_name(cs.FIELD_LEFT)) is not None + and _mentions(left, name) + for child in node.children + ) + + +def _body_reads(header: _Header, name: str) -> bool: + """Whether the body READS `name`. + + Through `_is_a_reference`, so `obj.a` and `helper(a=1)` do not count: + an attribute's name and a keyword argument's label are identifiers + spelled the same as the parameter but are not uses of it, and + refusing on those blocked legitimate removals (CodeRabbit, PR #1533). + """ + body = header.function.child_by_field_name(cs.FIELD_BODY) + if body is None: + return False + wanted = name.encode(cs.ENCODING_UTF8) + stack = [body] + while stack: + node = stack.pop() + if ( + node.type == cs.TS_PY_IDENTIFIER + and node.text == wanted + and _is_a_reference(node) + ): + return True + stack.extend(node.children) + return False + + +def _body_references( + header: _Header, old: str, new: str, renamed_away: Iterable[str] +) -> list[tuple[int, int]]: + """Byte spans of the parameter `old` in the body, to become `new`. + + Refuses when the name is used inside a nested scope or re-bound by a + statement, since the walk cannot tell those uses from the parameter's, + and when `new` is already read in the body, since the parameter would + then shadow it. A name that is itself being renamed away does not count + as taken: swapping two parameters is a rename in each direction. + """ + body = header.function.child_by_field_name(cs.FIELD_BODY) + assert body is not None + wanted = old.encode(cs.ENCODING_UTF8) + taken = None if new in renamed_away else new.encode(cs.ENCODING_UTF8) + rebinds = cs.SIGNATURE_BODY_REBINDS.format(old=old, qn=header.qn, new=new) + shadows = cs.SIGNATURE_BODY_NAME_TAKEN.format(old=old, qn=header.qn, new=new) + spans: list[tuple[int, int]] = [] + stack: list[Node] = [body] + while stack: + node = stack.pop() + if node.type in _OPAQUE_TO_THE_WALK: + _refuse_mentions(node, wanted, taken, rebinds, shadows) + continue + if node.type in _COMPREHENSIONS and _comprehension_binds(node, wanted): + raise SignatureRefused(rebinds) + if node.type == cs.TS_PY_IDENTIFIER and _is_a_reference(node): + if node.text == wanted: + spans.append((node.start_byte, node.end_byte)) + elif node.text == taken: + raise SignatureRefused(shadows) + stack.extend(node.children) + return spans + + +def _refuse_mentions( + node: Node, wanted: bytes, taken: bytes | None, rebinds: str, shadows: str +) -> None: + """A nested scope or re-binding statement may mention neither name. + + The walk cannot tell a use of the old name there from the parameter's; + and a nested scope reading the new name would start reading the renamed + parameter instead of whatever it read before. + """ + if _mentions(node, wanted): + raise SignatureRefused(rebinds) + if taken is not None and _mentions(node, taken): + raise SignatureRefused(shadows) + + +# --- call sites ----------------------------------------------------------------- + + +def _bind_arguments( + args: Node, source: bytes, old: Sequence[ParamSpec] +) -> dict[int, _Binding]: + """Old parameter index -> the value this site passes for it.""" + names = [p.name for p in old] + children = args.named_children + positional_kinds = { + cs.TS_PY_KEYWORD_ARGUMENT, + cs.TS_PY_LIST_SPLAT, + cs.TS_PY_DICTIONARY_SPLAT, + cs.TS_COMMENT, + } + given = sum(1 for child in children if child.type not in positional_kinds) + if given > len(old): + raise _Unmapped( + cs.SIGNATURE_SITE_TOO_MANY.format(given=given, declared=len(old)) + ) + bound: dict[int, _Binding] = {} + positional = 0 + for position, child in enumerate(children): + text = _text(child, source) + if child.type in ( + cs.TS_PY_LIST_SPLAT, + cs.TS_PY_DICTIONARY_SPLAT, + cs.TS_COMMENT, + ): + raise _Unmapped(cs.SIGNATURE_SITE_UNREADABLE.format(text=text)) + if child.type == cs.TS_PY_KEYWORD_ARGUMENT: + key = child.child_by_field_name(cs.FIELD_NAME) + value = child.child_by_field_name(cs.FIELD_VALUE) + assert key is not None and value is not None + name = _text(key, source) + if name not in names: + raise _Unmapped(cs.SIGNATURE_SITE_UNKNOWN_KEYWORD.format(name=name)) + index = names.index(name) + if index in bound: + raise _Unmapped(cs.SIGNATURE_SITE_DUPLICATE.format(name=name)) + bound[index] = _Binding( + _text(value, source), True, position, (value.start_byte, value.end_byte) + ) + continue + bound[positional] = _Binding( + text, False, position, (child.start_byte, child.end_byte) + ) + positional += 1 + return bound + + +def _fold( + candidate: _Candidate, edits: Sequence[_Edit] +) -> tuple[dict[int, _Binding], list[_Edit]]: + """The site's bindings with the planned edits inside them applied. + + A recursive call carries the body rename in its arguments; a call nested + in another call's arguments carries the inner site's rewrite. Folding + those into the value text lets the enclosing site be one edit, and the + folded edits are returned so the caller can drop them from the plan. + Any other edit overlapping the argument list cannot be folded, and the + site is left unmapped. + """ + path = candidate.site.path + start, end = candidate.span + inside = [ + edit + for edit in edits + if edit.path == path and edit.span[0] < end and start < edit.span[1] + ] + folded: dict[int, _Binding] = {} + consumed: list[_Edit] = [] + for index, binding in candidate.bound.items(): + lo, hi = binding.span + nested = [e for e in inside if lo <= e.span[0] and e.span[1] <= hi] + if not nested: + folded[index] = binding + continue + raw = bytearray(binding.text.encode(cs.ENCODING_UTF8)) + for edit in sorted(nested, key=lambda e: e.span[0], reverse=True): + raw[edit.span[0] - lo : edit.span[1] - lo] = edit.text.encode( + cs.ENCODING_UTF8 + ) + folded[index] = binding._replace(text=raw.decode(cs.ENCODING_UTF8)) + consumed.extend(nested) + if len(consumed) != len(inside): + raise _Unmapped(cs.SIGNATURE_SITE_OVERLAPS) + return folded, consumed + + +def _render_arguments( + new: Sequence[ParamSpec], + sources: Sequence[_Source | None], + bound: Mapping[int, _Binding], +) -> str: + """The site's new argument list, parenthesised. + + Positional values keep positional form in the new order until a value + goes by keyword or a defaulted parameter is left out; from there every + later value is spelled by keyword, since it can no longer sit in its + slot. Values that were keywords keep their original relative order + under their new names; values forced to keyword form follow in the new + parameter order. + """ + positional: list[str] = [] + keywords: list[tuple[tuple[int, int], str]] = [] + forced = False + for order, (spec, source) in enumerate(zip(new, sources, strict=True)): + found = _value_at_site(spec, source, bound) + if found is None: + forced = True + continue + value, binding = found + if binding is not None and binding.keyword: + forced = True + keywords.append(((0, binding.position), f"{spec.name}={value}")) + elif forced: + keywords.append(((1, order), f"{spec.name}={value}")) + else: + positional.append(value) + keywords.sort(key=lambda item: item[0]) + parts = positional + [text for _rank, text in keywords] + return f"({cs.SEPARATOR_COMMA_SPACE.join(parts)})" + + +def _value_at_site( + spec: ParamSpec, source: _Source | None, bound: Mapping[int, _Binding] +) -> tuple[str, _Binding | None] | None: + """The value this site passes for `spec`, with the binding it came from. + + None when the parameter's default stands in; unmapped when there is + neither a value nor a default. + """ + if source is not None and source.literal is not None: + return source.literal, None + if source is not None and source.index in bound: + binding = bound[source.index] + return binding.text, binding + if spec.has_default: + return None + raise _Unmapped(cs.SIGNATURE_SITE_NO_VALUE.format(name=spec.name)) + + +def _render_site( + candidate: _Candidate, + new: Sequence[ParamSpec], + sources: Sequence[_Source | None], + edits: list[_Edit], +) -> _Edit | None: + """The site's edit, with any edit planned inside its arguments folded in. + + The folded edits leave the plan only once the site renders, so a site + the mapping cannot complete keeps the inner rewrites as they were. + """ + bound, consumed = _fold(candidate, edits) + text = _render_arguments(new, sources, bound) + for edit in consumed: + edits.remove(edit) + if not consumed and text == candidate.text: + return None + return _Edit(candidate.site.path, candidate.span, text) + + +def _check_hierarchy( + qn: str, headers: Sequence[_Header], old_names: Sequence[str] +) -> None: + """Every override declares the same parameter names as the definition.""" + for header in headers[1:]: + theirs = [p.name for p in header.params] + if theirs != list(old_names): + raise SignatureRefused( + cs.SIGNATURE_HIERARCHY_MISMATCH.format( + qn=qn, + member=header.qn, + theirs=cs.SEPARATOR_COMMA_SPACE.join(theirs), + ours=cs.SEPARATOR_COMMA_SPACE.join(old_names), + ) + ) + + +def _definition_edits( + header: _Header, + new: Sequence[ParamSpec], + carried: set[str], + renamed: Sequence[tuple[str, str]], +) -> list[_Edit]: + """The header's new parameter list, and every body reference renamed.""" + own = {p.name: p for p in header.params} + texts = [own[spec.name].text if spec.name in carried else spec.text for spec in new] + if header.receiver is not None: + texts.insert(0, header.receiver) + edits = [ + _Edit(header.path, header.span, f"({cs.SEPARATOR_COMMA_SPACE.join(texts)})") + ] + renamed_away = {old for old, _new in renamed} + for old_name, new_name in renamed: + # A default is evaluated at DEFINITION time, in the enclosing scope, + # so `def f(a, b=a)` renamed a->x writes `def f(x, b=a)` and raises + # NameError the moment the module loads. `_body_references` walks the + # body only, so it cannot see the reference (Copilot, PR #1533). + _refuse_default_reading(header, texts, old_name, new_name) + edits.extend( + _Edit(header.path, span, new_name) + for span in _body_references(header, old_name, new_name, renamed_away) + ) + return edits + + +def _refuse_default_reading( + header: _Header, texts: Sequence[str], old: str, new: str +) -> None: + """Refuse when a parameter's default reads the name being renamed.""" + pattern = re.compile(rf"\b{re.escape(old)}\b") + for text in texts: + name, sep, default = text.partition(_LITERAL_PREFIX) + if not sep or not pattern.search(default): + continue + raise SignatureRefused( + cs.SIGNATURE_DEFAULT_REFERENCES_RENAMED.format( + old=old, qn=header.qn, new=new, param=name.strip() + ) + ) + + +def _params_of(params: Node, source: bytes, qn: str) -> list[ParamSpec]: + """Every parameter as a plain spec, or a refusal naming the odd one.""" + specs: list[ParamSpec] = [] + for child in params.named_children: + spec = _param_spec(child, source) + if spec is None: + raise SignatureRefused( + cs.SIGNATURE_UNSUPPORTED_PARAMS.format(qn=qn, text=_text(child, source)) + ) + specs.append(spec) + return specs + + +def _take_receiver( + qn: str, label: object, specs: list[ParamSpec], node: Node, source: bytes +) -> str | None: + """Pop a method's `self`/`cls` off `specs`; a static method has none. + + A method whose first parameter is named anything else refuses: `this` + would be remapped as a parameter and every bound call would lose its + receiver. + + Staticness is decided BEFORE the name: `@staticmethod def f(self, x)` + is legal Python and its `self` is an ordinary parameter, so popping it + by name dropped a real one from the header and every call site's + binding shifted by one (Copilot, PR #1912). + """ + if label != cs.NodeLabel.METHOD or not specs: + return None + if _is_static(node, source): + return None + if specs[0].name in cs.PY_RECEIVER_NAMES: + return specs.pop(0).text + raise SignatureRefused( + cs.SIGNATURE_UNUSUAL_RECEIVER.format(qn=qn, name=specs[0].name) + ) + + +# --- the operation --------------------------------------------------------------- + + +class SignatureChanger: + """Plan and apply one signature change against a project's graph.""" + + def __init__( + self, + repo_root: Path, + fetch_all: QueryFn, + project_name: str, + reingest: Reingest | None = None, + ) -> None: + self.repo_root = repo_root.resolve() + self.fetch_all = fetch_all + self.project = project_name + # With a re-ingest the change is held to its postcondition contract + # (issue #1531): the delta of what it wrote is measured and the + # transaction undone when the contract fails. + self.reingest = reingest + parsers, _queries = load_parsers() + self._parsers = dict(parsers) + self._parser: Parser = self._parsers[cs.SupportedLanguage.PYTHON] + + # -- planning -- + + def plan( + self, + qn: str, + new_params: Sequence[str], + mapping: Mapping[str, str] | None, + allow_heuristic: bool, + ) -> tuple[SignatureReport, list[_Edit]]: + """Everything the change touches, and the edits that do it.""" + patcher = Patcher(self.repo_root, parsers=self._parsers) + members = hierarchy(self.fetch_all, self.project, qn) + headers = [self._header(member, patcher) for member in members] + old = headers[0].params + old_names = [p.name for p in old] + _check_hierarchy(qn, headers, old_names) + new = _new_specs(new_params, old) + sources = _resolve_sources(new, old, mapping or {}) + for spec, source in zip(new, sources, strict=True): + if source is not None and source.literal is not None: + _check_literal(spec, source.literal) + # A bare old name carries that parameter's own spelling over (each + # override keeps its own); spelling a kept parameter out again + # re-annotates it. + # + # Read from the WRITTEN text, not from spec equality: a bare name + # resolves to the PRIMARY's spec, so an explicitly spelled `a: int` + # that happens to match the primary compared equal and counted as + # carried -- leaving an override declaring `a: str` un-annotated, + # which is the opposite of what was asked (Copilot, PR #1533). + carried = { + text.strip() + for text in new_params + if _IDENTIFIER_RE.fullmatch(text.strip()) and text.strip() in old_names + } + # An old parameter fed to a new one of another name is renamed, and + # the body must follow or the definition would be broken. + renamed = [ + (old_names[source.index], spec.name) + for spec, source in zip(new, sources, strict=True) + if source is not None + and source.index is not None + and old_names[source.index] != spec.name + ] + # A dropped parameter the body still reads leaves an unresolved + # name: `def f(a): return a` emptied writes `def f(): return a`, + # which PARSES and raises NameError only when called, so neither + # the syntax nor the arity postcondition catches it. Renamed names + # are excluded: the body rewrite follows those (Copilot, PR #1533). + kept = {spec.name for spec in new} | {old for old, _new in renamed} + for header in headers: + for dropped in (name for name in old_names if name not in kept): + if _body_reads(header, dropped): + raise SignatureRefused( + cs.SIGNATURE_DROPPED_STILL_READ.format( + name=dropped, qn=header.qn + ) + ) + sites: list[SignatureSite] = [] + edits: list[_Edit] = [] + for header in headers: + edits.extend(_definition_edits(header, new, carried, renamed)) + sites.append( + SignatureSite( + _DEFINITION, + header.path, + header.line, + header.col, + header.qn, + cs.EdgeResolution.EXACT, + ) + ) + unmapped = self._collect_sites( + headers, old, new, sources, allow_heuristic, patcher, sites, edits + ) + report = SignatureReport( + qualified_name=qn, + old_params=tuple(old_names), + new_params=tuple(spec.name for spec in new), + applied=False, + transaction_id="", + files=(), + sites=tuple(sites), + unmapped=tuple(unmapped), + hierarchy=tuple(members), + diff="", + message=cs.SIGNATURE_PLANNED.format( + count=sum(1 for s in sites if s.kind == _CALL), skipped=len(unmapped) + ), + ) + return report, edits + + def _header(self, qn: str, patcher: Patcher) -> _Header: + definition = graph_query.definition(self.fetch_all, self.project, qn, None) + path = definition["path"] + if not definition["found"] or not path: + raise SignatureRefused(cs.RENAME_UNKNOWN.format(qn=qn)) + if get_language_for_extension(Path(path).suffix) != cs.SupportedLanguage.PYTHON: + raise SignatureRefused( + cs.SIGNATURE_NOT_PYTHON.format(qn=qn, path=path, issue=LANGUAGES_ISSUE) + ) + try: + source = patcher.source(path) + except PatcherError as error: + raise SignatureRefused( + cs.SIGNATURE_DEFINITION_UNREADABLE.format(qn=qn, path=path, error=error) + ) from error + start = definition["start_line"] or 1 + end = definition["end_line"] or start + name = definition["name"] or qn.rsplit(cs.SEPARATOR_DOT, 1)[-1] + root = self._parser.parse(source).root_node + node = _find_definition(root, name, start, end) + params = node.child_by_field_name(cs.FIELD_PARAMETERS) if node else None + if node is None or params is None: + raise SignatureRefused(cs.SIGNATURE_NO_HEADER.format(qn=qn, path=path)) + specs = _params_of(params, source, qn) + receiver = _take_receiver(qn, definition["label"], specs, node, source) + return _Header( + qn, + path, + (params.start_byte, params.end_byte), + params.start_point[0] + 1, + params.start_point[1], + receiver, + specs, + node, + source, + ) + + def _collect_sites( + self, + headers: Sequence[_Header], + old: Sequence[ParamSpec], + new: Sequence[ParamSpec], + sources: Sequence[_Source | None], + allow_heuristic: bool, + patcher: Patcher, + sites: list[SignatureSite], + edits: list[_Edit], + ) -> list[UnmappedSite]: + unmapped: list[UnmappedSite] = [] + candidates = self._candidates(headers, old, allow_heuristic, patcher, unmapped) + # Innermost first, so a site nested in another's arguments (or a + # body rename inside a recursive call) is already planned when the + # enclosing site folds it into its value text. + order = sorted( + range(len(candidates)), + key=lambda i: ( + candidates[i].site.path, + candidates[i].span[1] - candidates[i].span[0], + candidates[i].span[0], + ), + ) + mapped: set[int] = set() + for index in order: + candidate = candidates[index] + try: + edit = _render_site(candidate, new, sources, edits) + except _Unmapped as skip: + unmapped.append( + UnmappedSite( + candidate.site.owner, + candidate.site.path, + candidate.site.line, + candidate.site.col, + str(skip), + ) + ) + continue + mapped.add(index) + # Into the plan at once: an enclosing site folds it in. + if edit is not None: + edits.append(edit) + # The report lists sites in the order the graph gave them. + sites.extend(c.site for i, c in enumerate(candidates) if i in mapped) + return unmapped + + def _candidates( + self, + headers: Sequence[_Header], + old: Sequence[ParamSpec], + allow_heuristic: bool, + patcher: Patcher, + unmapped: list[UnmappedSite], + ) -> list[_Candidate]: + """Every graph-known call site the mapping can read, once each.""" + candidates: list[_Candidate] = [] + seen: set[tuple[object, ...]] = set() + for header in headers: + for row in graph_query.callers(self.fetch_all, self.project, header.qn): + # Chained calls share a start and differ in their end, so + # the whole span identifies a site; sites without a + # location are each listed, since nothing tells them apart. + key = ( + row["path"], + row["line"], + row["col"], + row["end_line"], + row["end_col"], + ) + if None in key[:3] or key not in seen: + seen.add(key) + self._consider( + row, old, allow_heuristic, patcher, candidates, unmapped + ) + return candidates + + def _consider( + self, + row: graph_query.CallSiteRow, + old: Sequence[ParamSpec], + allow_heuristic: bool, + patcher: Patcher, + candidates: list[_Candidate], + unmapped: list[UnmappedSite], + ) -> None: + try: + candidates.append(self._candidate(row, old, allow_heuristic, patcher)) + except _Unmapped as skip: + unmapped.append( + UnmappedSite( + row["qualified_name"], + row["path"] or "", + row["line"], + row["col"], + str(skip), + ) + ) + + def _candidate( + self, + row: graph_query.CallSiteRow, + old: Sequence[ParamSpec], + allow_heuristic: bool, + patcher: Patcher, + ) -> _Candidate: + path, line, col = row["path"], row["line"], row["col"] + resolution = row["resolution"] + if path is None or line is None or col is None: + raise _Unmapped( + cs.SIGNATURE_SITE_NO_LOCATION.format( + resolution=resolution or cs.EdgeResolution.DYNAMIC.value + ) + ) + if resolution in AMBIGUOUS and not allow_heuristic: + raise _Unmapped(cs.SIGNATURE_SITE_GUESSED.format(resolution=resolution)) + if get_language_for_extension(Path(path).suffix) != cs.SupportedLanguage.PYTHON: + raise _Unmapped(cs.SIGNATURE_SITE_NOT_PYTHON.format(path=path)) + try: + source = patcher.source(path) + except PatcherError as error: + raise _Unmapped( + cs.SIGNATURE_SITE_UNREADABLE_FILE.format(error=error) + ) from error + end_line, end_col = row["end_line"], row["end_col"] + recorded_end = ( + (end_line - 1, end_col) + if end_line is not None and end_col is not None + else None + ) + root = self._parser.parse(source).root_node + call = call_node_at(root, line, col, recorded_end) + if call is None: + raise _Unmapped(cs.SIGNATURE_SITE_NO_CALL) + args = call.child_by_field_name(cs.FIELD_ARGUMENTS) + if args is None or args.type != cs.TS_ARGUMENT_LIST: + # `helper(x for x in xs)`: a generator, not a list of values. + raise _Unmapped( + cs.SIGNATURE_SITE_UNREADABLE.format(text=_text(args or call, source)) + ) + bound = _bind_arguments(args, source, old) + site = SignatureSite(_CALL, path, line, col, row["qualified_name"], resolution) + return _Candidate( + site, (args.start_byte, args.end_byte), _text(args, source), bound + ) + + # -- staging and applying -- + + def _stage( + self, edits: Iterable[_Edit] + ) -> tuple[EditTransaction, dict[str, object], list[str]]: + """Patch every edit into a transaction; nothing touches the tree.""" + patcher = Patcher(self.repo_root, parsers=self._parsers) + tx = EditTransaction(self.repo_root) + try: + for edit in edits: + patcher.replace_span(edit.path, edit.span, edit.text) + results = patcher.stage_into(tx) + except PatcherError as error: + # A plan the patcher cannot apply (an overlap the folding did + # not foresee) is a refusal, never a traceback past the tool. + tx.rollback() + raise SignatureRefused( + cs.SIGNATURE_STAGE_FAILED.format(error=error) + ) from error + broken = [key for key, result in results.items() if result.parses is False] + return tx, dict(results), broken + + def preview( + self, + qn: str, + new_params: Sequence[str], + mapping: Mapping[str, str] | None, + allow_heuristic: bool, + ) -> SignatureReport: + """Plan and stage, return the diff, and leave the tree untouched.""" + report, edits = self.plan(qn, new_params, mapping, allow_heuristic) + tx, results, broken = self._stage(edits) + try: + diff = tx.diff() + finally: + tx.rollback() + message = ( + cs.SIGNATURE_PARSE_FAILED.format( + files=cs.SEPARATOR_COMMA_SPACE.join(broken) + ) + if broken + else report.message + ) + return report._replace(files=tuple(sorted(results)), diff=diff, message=message) + + def apply( + self, + qn: str, + new_params: Sequence[str], + mapping: Mapping[str, str] | None, + allow_heuristic: bool, + ) -> SignatureReport: + """Plan, patch, verify and commit; the tree is untouched on failure.""" + report, edits = self.plan(qn, new_params, mapping, allow_heuristic) + tx, results, broken = self._stage(edits) + if broken: + tx.rollback() + return report._replace( + files=tuple(sorted(results)), + message=cs.SIGNATURE_PARSE_FAILED.format( + files=cs.SEPARATOR_COMMA_SPACE.join(broken) + ), + ) + outcome = tx.commit() + report = report._replace( + applied=outcome.applied, + transaction_id=outcome.transaction_id, + files=outcome.files, + diff=outcome.diff, + message=outcome.message, + ) + if outcome.applied and self.reingest is not None: + report = self._enforce_contract(report, allow_heuristic) + return report + + def _enforce_contract( + self, report: SignatureReport, allow_heuristic: bool + ) -> SignatureReport: + assert self.reingest is not None + try: + delta = measure( + self.fetch_all, + self.project, + self.repo_root, + report.files, + self.reingest, + ) + # The transaction has landed; a graph that cannot be measured is + # reported, never raised past the committed edit. + except Exception as error: # noqa: BLE001 + logger.warning(cs.SIGNATURE_CONTRACT_UNMEASURED.format(error=error)) + return report._replace( + verdict=None, + # `reingest` rejects bad paths with `ValueError` and converts + # every prologue failure to `ReingestAborted` before writing, + # so those two mean the graph was never touched (the rule + # `rename` and the guarded MCP callback apply). + graph_incomplete=not isinstance(error, ValueError | ReingestAborted), + message=cs.SIGNATURE_CONTRACT_UNMEASURED.format(error=error), + ) + verdict = verify( + change_signature_expectation( + [ + _site_key(site.path, site.line, site.col) + for site in report.unmapped + if site.line is not None + ], + heuristic_allowed=allow_heuristic, + ), + delta, + rewritten=[ + (_site_key(site.path, site.line, site.col), site.resolution) + for site in report.sites + if site.kind == _CALL + ], + ) + if verdict.ok: + return report._replace(verdict=verdict) + reasons = cs.SEPARATOR_SEMICOLON_SPACE.join(verdict.failures) + try: + # This change's own transaction, not whatever is newest: a later + # edit stacked on it refuses the rollback instead. + undo_transaction(self.repo_root, report.transaction_id) + except TransactionConflict as conflict: + logger.warning(str(conflict)) + return report._replace( + verdict=verdict, + message=cs.SIGNATURE_ROLLBACK_REFUSED.format(reasons=reasons), + ) + try: + self.reingest(list(report.files)) + # The files are restored; the graph may have lost the subtree the + # re-ingest deleted before failing. Say so, never raise. + except Exception as error: # noqa: BLE001 + logger.warning( + cs.SIGNATURE_ROLLBACK_UNMEASURED.format(reasons=reasons, error=error) + ) + return report._replace( + applied=False, + verdict=verdict, + graph_incomplete=True, + message=cs.SIGNATURE_ROLLBACK_UNMEASURED.format( + reasons=reasons, error=error + ), + ) + return report._replace( + applied=False, + verdict=verdict, + message=cs.SIGNATURE_CONTRACT_FAILED.format(reasons=reasons), + ) + + +def _site_key(path: str, line: int | None, col: int | None) -> str: + """The identity the contract checks a call under: two calls can share a line.""" + return cs.CHAR_COLON.join((path, str(line), str(col))) + + +def change_signature( + repo_root: Path, + fetch_all: QueryFn, + project_name: str, + qualified_name: str, + new_params: Sequence[str], + mapping: Mapping[str, str] | None = None, + allow_heuristic: bool = False, + dry_run: bool = False, + reingest: Reingest | None = None, +) -> SignatureReport: + """The op: plan (and refuse what cannot be stated) or plan and apply. + + With `reingest` the applied change is measured through the structural + delta and undone when its postcondition contract fails (issue #1531). + """ + changer = SignatureChanger(repo_root, fetch_all, project_name, reingest=reingest) + if dry_run: + return changer.preview(qualified_name, new_params, mapping, allow_heuristic) + return changer.apply(qualified_name, new_params, mapping, allow_heuristic) + + +def sites_for(sites: Iterable[SignatureSite]) -> list[dict[str, object]]: + return [dict(site._asdict()) for site in sites] + + +def unmapped_for(sites: Iterable[UnmappedSite]) -> list[dict[str, object]]: + return [dict(site._asdict()) for site in sites] diff --git a/codebase_rag/editing/sites.py b/codebase_rag/editing/sites.py new file mode 100644 index 0000000000..abbea9426d --- /dev/null +++ b/codebase_rag/editing/sites.py @@ -0,0 +1,72 @@ +"""Graph-site helpers the edit operations share. + +`rename` (issue #1532) and `change_signature` (issue #1533) both walk a +method's override hierarchy and both need the call node a recorded site +refers to; keeping one copy means the two operations cannot disagree about +which call a `(line, col)` names. +""" + +from __future__ import annotations + +from tree_sitter import Node + +from .. import constants as cs +from .. import graph_query +from ..graph_query import QueryFn + +# Resolutions that bound a site by guesswork: an operation rewrites through +# them only when the caller accepts the risk with `allow_heuristic`. +AMBIGUOUS = frozenset( + { + cs.EdgeResolution.HEURISTIC.value, + cs.EdgeResolution.OVERLOAD.value, + cs.EdgeResolution.DYNAMIC.value, + } +) + + +def hierarchy(fetch_all: QueryFn, project: str, qn: str) -> list[str]: + """`qn` plus every method it overrides or is overridden by, transitively.""" + seen: list[str] = [qn] + frontier = [qn] + while frontier: + current = frontier.pop() + for row in graph_query.overrides(fetch_all, project, current): + other = row["qualified_name"] + if other not in seen: + seen.append(other) + frontier.append(other) + return seen + + +def call_node_at( + root: Node, line: int, col: int, recorded_end: tuple[int, int] | None +) -> Node | None: + """The call node at (line, col) that the graph site refers to. + + Several calls can share a start point -- `helper(helper(1))`, + `helper(2).upper()`, and both links of `obj.helper(1).helper(2)` -- so the + right one is the call ending where the site recorded its end, or the + outermost when no end was recorded. A recorded end that matches no call + names nothing: the position is stale, and rewriting a neighbour that + happens to share the start would be a guess. + """ + calls = _calls_starting_at(root, line - 1, col) + if recorded_end is not None: + return next((call for call in calls if call.end_point == recorded_end), None) + return max(calls, key=lambda call: call.end_byte, default=None) + + +def _calls_starting_at(root: Node, row: int, col: int) -> list[Node]: + calls: list[Node] = [] + stack: list[Node] = [root] + while stack: + node = stack.pop() + if ( + node.start_point == (row, col) + and node.child_by_field_name(cs.FIELD_FUNCTION) is not None + ): + calls.append(node) + if node.start_point[0] <= row <= node.end_point[0]: + stack.extend(node.children) + return calls diff --git a/codebase_rag/mcp/tools.py b/codebase_rag/mcp/tools.py index b84b341eb3..b319abc45e 100644 --- a/codebase_rag/mcp/tools.py +++ b/codebase_rag/mcp/tools.py @@ -120,6 +120,7 @@ def _read_file_slice(full_path: Path, start: int, limit: int | None) -> str: cs.MCPToolName.UPDATE_REPOSITORY, cs.MCPToolName.REINGEST, cs.MCPToolName.RENAME, + cs.MCPToolName.CHANGE_SIGNATURE, cs.MCPToolName.SURGICAL_REPLACE_CODE, cs.MCPToolName.READ_FILE, cs.MCPToolName.WRITE_FILE, @@ -486,6 +487,7 @@ def __init__( self.tests_reaching, ), cs.MCPToolName.RENAME: self._rename_tool(), + cs.MCPToolName.CHANGE_SIGNATURE: self._change_signature_tool(), cs.MCPToolName.QUERY_CODE_GRAPH: ToolMetadata( name=cs.MCPToolName.QUERY_CODE_GRAPH, description=td.MCP_TOOLS[cs.MCPToolName.QUERY_CODE_GRAPH], @@ -2766,6 +2768,119 @@ def _run_rename( payload[cs.DICT_KEY_ERROR] = payload_marker_error return payload + def _change_signature_tool(self) -> ToolMetadata: + def prop(kind: cs.MCPSchemaType, description: str) -> MCPInputSchemaProperty: + return MCPInputSchemaProperty(type=kind, description=description) + + return ToolMetadata( + name=cs.MCPToolName.CHANGE_SIGNATURE, + description=td.MCP_TOOLS[cs.MCPToolName.CHANGE_SIGNATURE], + input_schema=MCPInputSchema( + type=cs.MCPSchemaType.OBJECT, + properties={ + cs.MCPParamName.QUALIFIED_NAME: prop( + cs.MCPSchemaType.STRING, td.MCP_PARAM_QUALIFIED_NAME + ), + cs.MCPParamName.NEW_PARAMS: MCPInputSchemaProperty( + type=cs.MCPSchemaType.ARRAY, + description=td.MCP_PARAM_NEW_PARAMS, + items={cs.MCPSchemaField.TYPE: cs.MCPSchemaType.STRING}, + ), + cs.MCPParamName.MAPPING: prop( + cs.MCPSchemaType.OBJECT, td.MCP_PARAM_MAPPING + ), + cs.MCPParamName.ALLOW_HEURISTIC: prop( + cs.MCPSchemaType.BOOLEAN, td.MCP_PARAM_ALLOW_HEURISTIC + ), + cs.MCPParamName.DRY_RUN: prop( + cs.MCPSchemaType.BOOLEAN, td.MCP_PARAM_RENAME_DRY_RUN + ), + cs.MCPParamName.PROJECT: prop( + cs.MCPSchemaType.STRING, td.MCP_PARAM_PROJECT + ), + }, + required=[ + cs.MCPParamName.QUALIFIED_NAME, + cs.MCPParamName.NEW_PARAMS, + ], + ), + handler=self.change_signature, + returns_json=True, + ) + + async def change_signature( + self, + qualified_name: str, + new_params: list[str], + mapping: dict[str, str] | None = None, + allow_heuristic: bool = False, + dry_run: bool = False, + project: str | None = None, + ) -> object: + # Under the ingestor lock for the same reason as `rename`: a rebuild + # landing between planning and writing would rewrite stale sites. + return await self._graph_query( + cs.MCPToolName.CHANGE_SIGNATURE, + project, + lambda name: self._run_change_signature( + name, qualified_name, new_params, mapping, allow_heuristic, dry_run + ), + ) + + def _run_change_signature( + self, + project_name: str, + qualified_name: str, + new_params: list[str], + mapping: dict[str, str] | None, + allow_heuristic: bool, + dry_run: bool, + ) -> object: + from codebase_rag.editing.signature import ( + SignatureRefused, + change_signature, + sites_for, + unmapped_for, + ) + + # Paths are meaningful only under the root the project was indexed + # from (issue #1542), exactly as for `rename`. + root = graph_query.source_root_for( + self.ingestor.fetch_all, project_name, Path(self.project_root) + ) + if root is None: + return { + cs.DICT_KEY_ERROR: cs.RENAME_WRONG_ROOT.format(project=project_name) + } + try: + report = change_signature( + root, + self.ingestor.fetch_all, + project_name, + qualified_name, + new_params, + mapping, + allow_heuristic=allow_heuristic, + dry_run=dry_run, + reingest=self._guarded_rename_reingest(project_name), + ) + except SignatureRefused as refused: + return {cs.DICT_KEY_ERROR: str(refused)} + payload_marker_error: str | None = None + if report.graph_incomplete: + # The rollback's re-ingest failed: the same invalidation, in + # memory and durably, that `_run_rename` applies. + self._live_updater = None + self._invalidate_graph_for(project_name) + payload_marker_error = self._require_marker(project_name, writing=True) + payload = dict(report._asdict()) + payload[cs.KEY_SITES] = sites_for(report.sites) + payload[cs.KEY_UNMAPPED] = unmapped_for(report.unmapped) + payload[cs.KEY_VERDICT] = report.verdict._asdict() if report.verdict else None + if payload_marker_error is not None: + payload[cs.DICT_KEY_ERROR] = payload_marker_error + return payload + async def query_code_graph( self, natural_language_query: str, project: str | None = None ) -> QueryResultDict: diff --git a/codebase_rag/tests/test_change_signature.py b/codebase_rag/tests/test_change_signature.py new file mode 100644 index 0000000000..02970d9c5f --- /dev/null +++ b/codebase_rag/tests/test_change_signature.py @@ -0,0 +1,1521 @@ +# Edit algebra op 2, change_signature (issue #1533): the definition (and its +# override hierarchy) gets the new parameter list, and every graph-known call +# site is rewritten per an explicit mapping from new parameter to old value. +# Sites the mapping cannot complete, and sites the graph resolved by +# guesswork, are left as written and listed as unmapped. The graph is the +# in-memory stateful ingestor over a real index of a fixture repo. +from __future__ import annotations + +import subprocess +import sys +from pathlib import Path + +import pytest + +from codebase_rag import constants as cs +from codebase_rag.editing.signature import ( + SignatureRefused, + change_signature, + parse_mapping, +) +from codebase_rag.editing.transaction import load_history +from codebase_rag.graph_updater import GraphUpdater +from codebase_rag.parser_loader import load_parsers +from evals.cgr_graph import _StatefulIngestor + +PROJECT = "signature_fixture" +FIXTURE: dict[str, str] = { + "pkg/__init__.py": "", + "pkg/util.py": "def helper(a: int, b: str = 'x') -> str:\n return b * a\n", + "pkg/app.py": ( + "from pkg.util import helper\n\n\n" + "def run():\n return helper(2)\n\n\n" + "def run_kw():\n return helper(2, b='y')\n\n\n" + "def run_both():\n return helper(3, 'z')\n" + ), + "tests/__init__.py": "", + "tests/test_app.py": ( + "from pkg.app import run\n\n\ndef test_run():\n assert run() == 'xx'\n" + ), +} +HELPER = f"{PROJECT}.pkg.util.helper" + + +def _write(root: Path, rel: str, text: str) -> None: + path = root / rel + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + + +def _index(root: Path) -> tuple[_StatefulIngestor, GraphUpdater]: + parsers, queries = load_parsers() + store = _StatefulIngestor() + updater = GraphUpdater( + ingestor=store, + repo_path=root, + parsers=parsers, + queries=queries, + project_name=PROJECT, + ) + updater.run(force=True) + return store, updater + + +def _project(temp_repo: Path, files: dict[str, str]) -> Path: + root = temp_repo / PROJECT + root.mkdir() + for rel, text in files.items(): + _write(root, rel, text) + return root + + +@pytest.fixture +def repo(temp_repo: Path) -> tuple[Path, _StatefulIngestor, GraphUpdater]: + root = _project(temp_repo, FIXTURE) + store, updater = _index(root) + return root, store, updater + + +def _read(root: Path, rel: str) -> str: + return (root / rel).read_text(encoding="utf-8") + + +def _smoke(root: Path, expression: str) -> None: + """Run the rewritten fixture for real: a rewrite that parses can still + be wrong, and only executing the call sites against the new definition + shows they agree.""" + result = subprocess.run( + [sys.executable, "-c", expression], + cwd=root, + capture_output=True, + text=True, + encoding="utf-8", + check=False, + ) + assert result.returncode == 0, result.stderr + + +def _set_resolution(store: _StatefulIngestor, caller: str, resolution: str) -> None: + edge = next( + e + for e in store.edges + if e[1] == caller and e[2] == cs.RelationshipType.CALLS.value and e[4] == HELPER + ) + store.edge_props[edge][cs.KEY_RESOLUTION] = resolution + + +# --- acceptance (issue #1533) -------------------------------------------------- + + +def test_add_required_parameter_with_a_default_mapping( + repo: tuple[Path, _StatefulIngestor, GraphUpdater], +) -> None: + root, store, updater = repo + report = change_signature( + root, + store.fetch_all, + PROJECT, + HELPER, + ["a", "n: int", "b"], + {"n": "=1"}, + reingest=updater.reingest, + ) + assert report.applied, report.message + assert report.old_params == ("a", "b") + assert report.new_params == ("a", "n", "b") + # The new parameter is REQUIRED in the definition: the literal is what + # existing sites pass, not a default the definition gains. + assert "def helper(a: int, n: int, b: str = 'x') -> str:" in _read( + root, "pkg/util.py" + ) + app = _read(root, "pkg/app.py") + assert "return helper(2, 1)" in app + assert "return helper(2, 1, b='y')" in app + assert "return helper(3, 1, 'z')" in app + assert report.unmapped == () + assert report.verdict is not None, report.message + assert report.verdict.ok, report.message + assert [t["qualified_name"] for t in report.verdict.affected_tests] == [ + f"{PROJECT}.tests.test_app.test_run" + ] + assert set(report.files) == {"pkg/util.py", "pkg/app.py"} + _smoke( + root, + "from pkg.app import run, run_kw, run_both\n" + "assert (run(), run_kw(), run_both()) == ('xx', 'yy', 'zzz')\n", + ) + + +def test_reorder_parameters_rewrites_positional_callers_only( + temp_repo: Path, +) -> None: + root = _project( + temp_repo, + { + "pkg/__init__.py": "", + "pkg/util.py": "def helper(a: int, b: str) -> str:\n return b * a\n", + "pkg/app.py": ( + "from pkg.util import helper\n\n\n" + "def run():\n return helper(2, 'x')\n\n\n" + "def run_kw():\n return helper(a=2, b='y')\n" + ), + }, + ) + store, updater = _index(root) + before_kw = "return helper(a=2, b='y')" + report = change_signature( + root, + store.fetch_all, + PROJECT, + HELPER, + ["b", "a"], + reingest=updater.reingest, + ) + assert report.applied, report.message + # A bare old name carries the old annotation and default over. + assert "def helper(b: str, a: int) -> str:" in _read(root, "pkg/util.py") + app = _read(root, "pkg/app.py") + assert "return helper('x', 2)" in app + # A keyword caller binds by name and is left exactly as written. + assert before_kw in app + assert report.verdict is not None, report.message + assert report.verdict.ok, report.message + _smoke( + root, + "from pkg.app import run, run_kw\nassert (run(), run_kw()) == ('xx', 'yy')", + ) + + +def test_heuristic_site_is_listed_as_unmapped_not_rewritten( + repo: tuple[Path, _StatefulIngestor, GraphUpdater], +) -> None: + root, store, _updater = repo + _set_resolution( + store, f"{PROJECT}.pkg.app.run_both", cs.EdgeResolution.HEURISTIC.value + ) + report = change_signature( + root, + store.fetch_all, + PROJECT, + HELPER, + ["a", "n: int", "b"], + {"n": "=1"}, + dry_run=True, + ) + (skipped,) = report.unmapped + assert skipped.owner == f"{PROJECT}.pkg.app.run_both" + assert skipped.path == "pkg/app.py" + assert cs.EdgeResolution.HEURISTIC.value in skipped.reason + assert [s.owner for s in report.sites if s.kind == "call"] == [ + f"{PROJECT}.pkg.app.run", + f"{PROJECT}.pkg.app.run_kw", + ] + # The diff carries the two rewritten sites and not the guessed one. + assert "+ return helper(2, 1)" in report.diff + assert "helper(3, 1, 'z')" not in report.diff + assert _read(root, "pkg/app.py") == FIXTURE["pkg/app.py"] + + +def test_allow_heuristic_rewrites_the_guessed_site( + repo: tuple[Path, _StatefulIngestor, GraphUpdater], +) -> None: + root, store, _updater = repo + _set_resolution( + store, f"{PROJECT}.pkg.app.run_both", cs.EdgeResolution.HEURISTIC.value + ) + report = change_signature( + root, + store.fetch_all, + PROJECT, + HELPER, + ["a", "n: int", "b"], + {"n": "=1"}, + allow_heuristic=True, + dry_run=True, + ) + assert report.unmapped == () + assert "helper(3, 1, 'z')" in report.diff + + +def test_allow_heuristic_applies_through_the_contract( + repo: tuple[Path, _StatefulIngestor, GraphUpdater], +) -> None: + # The contract refuses a rewritten guessed site unless the operation + # was told to allow it; the leave must reach the expectation. + root, store, updater = repo + _set_resolution( + store, f"{PROJECT}.pkg.app.run_both", cs.EdgeResolution.HEURISTIC.value + ) + report = change_signature( + root, + store.fetch_all, + PROJECT, + HELPER, + ["a", "n: int", "b"], + {"n": "=1"}, + allow_heuristic=True, + reingest=updater.reingest, + ) + assert report.applied, report.message + assert report.verdict is not None, report.message + assert report.verdict.ok, report.message + assert "return helper(3, 1, 'z')" in _read(root, "pkg/app.py") + + +# --- the mapping ------------------------------------------------------------------- + + +def test_a_parameter_renamed_by_index_renames_keyword_sites(temp_repo: Path) -> None: + root = _project( + temp_repo, + { + "pkg/__init__.py": "", + "pkg/util.py": "def helper(a: int, b: str) -> str:\n return b * a\n", + "pkg/app.py": ( + "from pkg.util import helper\n\n\n" + "def run():\n return helper(2, 'x')\n\n\n" + "def run_kw():\n return helper(b='y', a=2)\n" + ), + }, + ) + store, updater = _index(root) + report = change_signature( + root, + store.fetch_all, + PROJECT, + HELPER, + ["times: int", "text: str"], + {"times": "0", "text": "1"}, + reingest=updater.reingest, + ) + assert report.applied, report.message + util = _read(root, "pkg/util.py") + assert "def helper(times: int, text: str) -> str:" in util + # The body follows the rename, or the definition would be broken. + assert " return text * times\n" in util + app = _read(root, "pkg/app.py") + assert "return helper(2, 'x')" in app + # Keyword values keep their order and form; only the names follow. + assert "return helper(text='y', times=2)" in app + assert report.verdict is not None, report.message + assert report.verdict.ok, report.message + _smoke( + root, + "from pkg.app import run, run_kw\nassert (run(), run_kw()) == ('xx', 'yy')", + ) + + +def test_a_value_after_an_omitted_default_is_spelled_by_keyword( + repo: tuple[Path, _StatefulIngestor, GraphUpdater], +) -> None: + root, store, updater = repo + report = change_signature( + root, + store.fetch_all, + PROJECT, + HELPER, + ["a", "b", "n: int = 0"], + {"n": "=1"}, + reingest=updater.reingest, + ) + assert report.applied, report.message + app = _read(root, "pkg/app.py") + # `helper(2)` relied on b's default; the value for n cannot sit in b's + # slot, so it goes by keyword. + assert "return helper(2, n=1)" in app + assert "return helper(2, b='y', n=1)" in app + assert "return helper(3, 'z', 1)" in app + _smoke( + root, + "from pkg.app import run, run_kw, run_both\n" + "assert (run(), run_kw(), run_both()) == ('xx', 'yy', 'zzz')\n", + ) + + +def test_an_unmapped_required_parameter_leaves_every_site_and_lists_it( + repo: tuple[Path, _StatefulIngestor, GraphUpdater], +) -> None: + root, store, updater = repo + report = change_signature( + root, + store.fetch_all, + PROJECT, + HELPER, + ["a", "n: int", "b"], + reingest=updater.reingest, + ) + assert report.applied, report.message + assert "def helper(a: int, n: int, b: str = 'x') -> str:" in _read( + root, "pkg/util.py" + ) + assert _read(root, "pkg/app.py") == FIXTURE["pkg/app.py"] + assert sorted((u.path, u.line) for u in report.unmapped) == [ + ("pkg/app.py", 5), + ("pkg/app.py", 9), + ("pkg/app.py", 13), + ] + assert all("`n`" in u.reason for u in report.unmapped) + # Listed sites satisfy the contract: the operation said it left them. + assert report.verdict is not None, report.message + assert report.verdict.ok, report.message + + +def _strip_location(store: _StatefulIngestor, caller: str) -> None: + edge = next( + e + for e in store.edges + if e[1] == caller and e[2] == cs.RelationshipType.CALLS.value and e[4] == HELPER + ) + props = store.edge_props[edge] + props[cs.KEY_RESOLUTION] = cs.EdgeResolution.DYNAMIC.value + for key in (cs.KEY_LINE, cs.KEY_COL, cs.KEY_END_LINE, cs.KEY_END_COL): + props.pop(key, None) + + +def test_a_site_whose_recorded_end_matches_no_call_is_listed( + repo: tuple[Path, _StatefulIngestor, GraphUpdater], +) -> None: + # A stale end position must not fall back to another call sharing the + # start: the site is listed rather than the wrong call rewritten. + root, store, _updater = repo + edge = next( + e + for e in store.edges + if e[1] == f"{PROJECT}.pkg.app.run_kw" + and e[2] == cs.RelationshipType.CALLS.value + and e[4] == HELPER + ) + store.edge_props[edge][cs.KEY_END_COL] = 99 + report = change_signature( + root, + store.fetch_all, + PROJECT, + HELPER, + ["a", "n: int", "b"], + {"n": "=1"}, + dry_run=True, + ) + (skipped,) = report.unmapped + assert skipped.owner == f"{PROJECT}.pkg.app.run_kw" + assert "no call" in skipped.reason + assert "helper(2, b='y')" not in report.diff + assert "+ return helper(3, 1, 'z')" in report.diff + + +def test_chained_calls_sharing_a_start_are_both_rewritten(temp_repo: Path) -> None: + # `B().area(2).area(3)`: both calls start where `B` does and differ only + # in where they end. The graph records both; neither may be dropped. + root = _project( + temp_repo, + { + "pkg/__init__.py": "", + "pkg/shapes.py": ( + "class B:\n" + " def area(self, scale):\n" + " return self\n\n\n" + "def run():\n" + " return B().area(2).area(3)\n" + ), + }, + ) + store, updater = _index(root) + run_qn = f"{PROJECT}.pkg.shapes.run" + stored = store.fetch_all + + def both_links(query: str, params: dict[str, object] | None = None): # type: ignore[no-untyped-def] + out = [] + for row in stored(query, params): + if row.get(cs.KEY_QUALIFIED_NAME) != run_qn: + out.append(row) + continue + for end_col in (22, 30): + link = dict(row) + link[cs.KEY_LINE] = 7 + link[cs.KEY_COL] = 11 + link[cs.KEY_END_LINE] = 7 + link[cs.KEY_END_COL] = end_col + out.append(link) + return out + + report = change_signature( + root, + both_links, + PROJECT, + f"{PROJECT}.pkg.shapes.B.area", + ["scale", "unit: str"], + {"unit": "='m'"}, + reingest=updater.reingest, + ) + assert report.applied, report.message + assert report.unmapped == () + assert sum(1 for s in report.sites if s.kind == "call") == 2 + assert "return B().area(2, 'm').area(3, 'm')" in _read(root, "pkg/shapes.py") + _smoke(root, "from pkg.shapes import run, B\nassert isinstance(run(), B)") + + +def test_every_site_without_a_location_is_listed( + repo: tuple[Path, _StatefulIngestor, GraphUpdater], +) -> None: + root, store, _updater = repo + _strip_location(store, f"{PROJECT}.pkg.app.run") + _strip_location(store, f"{PROJECT}.pkg.app.run_kw") + report = change_signature( + root, + store.fetch_all, + PROJECT, + HELPER, + ["a", "n: int", "b"], + {"n": "=1"}, + dry_run=True, + ) + assert sorted(u.owner for u in report.unmapped) == [ + f"{PROJECT}.pkg.app.run", + f"{PROJECT}.pkg.app.run_kw", + ] + assert all("no location" in u.reason for u in report.unmapped) + assert "+ return helper(3, 1, 'z')" in report.diff + + +def test_a_site_passing_surplus_arguments_is_listed_not_truncated( + temp_repo: Path, +) -> None: + root = _project( + temp_repo, + { + "pkg/__init__.py": "", + "pkg/util.py": "def helper(a, b):\n return a + b\n", + "pkg/app.py": ( + "from pkg.util import helper\n\n\n" + "def run():\n return helper(1, 2)\n\n\n" + "def broken():\n return helper(1, 2, 3)\n" + ), + }, + ) + store, updater = _index(root) + report = change_signature( + root, + store.fetch_all, + PROJECT, + HELPER, + ["b", "a"], + reingest=updater.reingest, + ) + assert report.applied, report.message + app = _read(root, "pkg/app.py") + assert "return helper(2, 1)" in app + assert "return helper(1, 2, 3)" in app + (skipped,) = report.unmapped + assert (skipped.path, skipped.line) == ("pkg/app.py", 9) + assert "3" in skipped.reason + assert report.verdict is not None, report.message + assert report.verdict.ok, report.message + + +@pytest.mark.parametrize( + ("site", "fragment"), + [ + ("helper(*args)", "*args"), + ("helper(2, **opts)", "**opts"), + ("helper(2, c=1)", "c"), + ], +) +def test_a_site_the_mapping_cannot_read_is_listed( + temp_repo: Path, site: str, fragment: str +) -> None: + root = _project( + temp_repo, + { + "pkg/__init__.py": "", + "pkg/util.py": FIXTURE["pkg/util.py"], + "pkg/app.py": ( + "from pkg.util import helper\n\n\n" + f"def run(args=(), opts={{}}):\n return {site}\n" + ), + }, + ) + store, _updater = _index(root) + report = change_signature( + root, + store.fetch_all, + PROJECT, + HELPER, + ["a", "n: int", "b"], + {"n": "=1"}, + dry_run=True, + ) + (skipped,) = report.unmapped + assert (skipped.path, skipped.line) == ("pkg/app.py", 5) + assert fragment in skipped.reason + assert site not in report.diff + + +@pytest.mark.parametrize( + ("mapping", "reason"), + [ + ({"n": "c"}, r"neither an old parameter"), # no old parameter c + ({"n": "2"}, r"only 2 old parameter"), # only two old parameters + ({"m": "=1"}, r"not a new parameter"), # no new parameter m + # One old value cannot feed two parameters. Matched by reason: with + # the duplicate check removed this case was still refused, by the + # body check seeing `a` already in use, so a bare `raises` could + # not tell the two apart. + ({"a": "b", "b": "b"}, r"cannot feed both"), + ], +) +def test_a_mapping_that_names_nothing_is_refused( + repo: tuple[Path, _StatefulIngestor, GraphUpdater], + mapping: dict[str, str], + reason: str, +) -> None: + root, store, _updater = repo + with pytest.raises(SignatureRefused, match=reason): + change_signature( + root, + store.fetch_all, + PROJECT, + HELPER, + ["a", "n: int", "b"], + mapping, + dry_run=True, + ) + assert _read(root, "pkg/util.py") == FIXTURE["pkg/util.py"] + + +def test_parse_mapping_text_forms() -> None: + assert parse_mapping(["n==1", "a=0", "b=old_b", "s=='x'"]) == { + "n": "=1", + "a": "0", + "b": "old_b", + "s": "='x'", + } + with pytest.raises(SignatureRefused): + parse_mapping(["n"]) + + +# --- the definition ------------------------------------------------------------------ + + +def test_default_literal_incompatible_with_the_declared_type_is_refused( + repo: tuple[Path, _StatefulIngestor, GraphUpdater], +) -> None: + root, store, _updater = repo + with pytest.raises(SignatureRefused, match=r"does not fit"): + change_signature( + root, + store.fetch_all, + PROJECT, + HELPER, + ["a", "n: int", "b"], + {"n": "='no'"}, + dry_run=True, + ) + # The old annotation counts too: `b: str` is carried over by the bare name. + with pytest.raises(SignatureRefused, match=r"does not fit"): + change_signature( + root, + store.fetch_all, + PROJECT, + HELPER, + ["a", "b"], + {"b": "=3"}, + dry_run=True, + ) + # An optional annotation admits None; a non-literal is not checked. + for annotation, literal in ( + ("int | None", "=None"), + ("Optional[int]", "=None"), + ("int", "=LIMIT"), + ("Sequence[int]", "=()"), + ): + report = change_signature( + root, + store.fetch_all, + PROJECT, + HELPER, + ["a", f"n: {annotation}", "b"], + {"n": literal}, + dry_run=True, + ) + assert report.unmapped == () + + +def test_a_kept_parameter_spelled_anew_takes_the_new_spelling( + repo: tuple[Path, _StatefulIngestor, GraphUpdater], +) -> None: + # `b` exists already; spelling it out again re-annotates it. Only the + # bare name carries the old spelling over. + root, store, _updater = repo + report = change_signature( + root, + store.fetch_all, + PROJECT, + HELPER, + ["a", "b: int = 0"], + dry_run=True, + ) + assert "+def helper(a: int, b: int = 0) -> str:" in report.diff + # Sites keep their values: the parameters did not move. + assert "helper(2, b='y')" not in report.diff + # The new annotation is the one a literal is checked against. + with pytest.raises(SignatureRefused, match=r"does not fit"): + change_signature( + root, + store.fetch_all, + PROJECT, + HELPER, + ["a", "b: int = 0"], + {"b": "='q'"}, + dry_run=True, + ) + + +def test_a_required_parameter_after_a_default_is_refused( + repo: tuple[Path, _StatefulIngestor, GraphUpdater], +) -> None: + root, store, _updater = repo + with pytest.raises(SignatureRefused, match=r"after"): + change_signature( + root, + store.fetch_all, + PROJECT, + HELPER, + ["a", "b", "n: int"], + {"n": "=1"}, + ) + assert _read(root, "pkg/util.py") == FIXTURE["pkg/util.py"] + assert load_history(root) == [] + + +@pytest.mark.parametrize( + "header", + [ + "def helper(a, *rest):", + "def helper(a, **extra):", + "def helper(a, /, b):", + "def helper(a, *, b):", + ], +) +def test_a_definition_with_special_parameters_is_refused( + temp_repo: Path, header: str +) -> None: + root = _project( + temp_repo, + { + "pkg/__init__.py": "", + "pkg/util.py": f"{header}\n return a\n", + "pkg/app.py": "from pkg.util import helper\n\n\ndef run():\n return helper(1)\n", + }, + ) + store, _updater = _index(root) + with pytest.raises(SignatureRefused, match=r"positional-or-keyword"): + change_signature( + root, store.fetch_all, PROJECT, HELPER, ["a", "n"], dry_run=True + ) + + +def test_a_non_python_definition_is_refused(temp_repo: Path) -> None: + root = _project( + temp_repo, + { + "lib/tool.js": "export function helper(a) { return a; }\n", + "lib/app.js": "import { helper } from './tool.js';\nexport const run = () => helper(1);\n", + }, + ) + store, _updater = _index(root) + with pytest.raises(SignatureRefused, match=r"Python"): + change_signature( + root, store.fetch_all, PROJECT, f"{PROJECT}.lib.tool.helper", ["a", "n"] + ) + + +def test_an_unknown_definition_is_refused( + repo: tuple[Path, _StatefulIngestor, GraphUpdater], +) -> None: + root, store, _updater = repo + with pytest.raises(SignatureRefused, match=r"No definition"): + change_signature( + root, store.fetch_all, PROJECT, f"{PROJECT}.pkg.util.nothing", ["a"] + ) + + +# --- renamed parameters in the body ---------------------------------------------- + + +def _rename_dry_run(temp_repo: Path, body: str) -> str: + root = _project( + temp_repo, + { + "pkg/__init__.py": "", + "pkg/util.py": f"def helper(a, b):\n{body}", + "pkg/app.py": "from pkg.util import helper\n\n\ndef run():\n return helper(2, 'x')\n", + }, + ) + store, _updater = _index(root) + report = change_signature( + root, + store.fetch_all, + PROJECT, + HELPER, + ["times", "text"], + {"times": "a", "text": "b"}, + dry_run=True, + ) + return report.diff + + +def test_a_renamed_parameter_is_renamed_where_the_body_reads_it( + temp_repo: Path, +) -> None: + diff = _rename_dry_run( + temp_repo, + # A comprehension reads the parameter without re-binding it; an + # attribute and a keyword argument merely share its spelling. + " total = [i * a for i in range(len(b))]\n" + " a = a + 1\n" + " return b.a(a=a, text=b) or total\n", + ) + assert "+def helper(times, text):" in diff + assert "+ total = [i * times for i in range(len(text))]" in diff + assert "+ times = times + 1" in diff + assert "+ return text.a(a=times, text=text) or total" in diff + + +def test_swapping_two_parameter_names_renames_both_ways(temp_repo: Path) -> None: + root = _project( + temp_repo, + { + "pkg/__init__.py": "", + "pkg/util.py": "def helper(a, b):\n return b * a\n", + "pkg/app.py": "from pkg.util import helper\n\n\ndef run():\n return helper(2, 'x')\n", + }, + ) + store, _updater = _index(root) + report = change_signature( + root, + store.fetch_all, + PROJECT, + HELPER, + ["b", "a"], + {"b": "a", "a": "b"}, + dry_run=True, + ) + assert "+def helper(b, a):" in report.diff + assert "+ return a * b" in report.diff + # The values stay where they were: only the names moved. + assert "helper(2, 'x')" not in report.diff + + +@pytest.mark.parametrize( + "body", + [ + " def inner():\n return a\n return inner() * b\n", + " return (lambda: a)() * b\n", + " return [a for a in range(len(b))]\n", + " global a\n return b\n", + " from pkg import a\n return b\n", + ], +) +def test_a_parameter_used_in_a_nested_scope_refuses_the_rename( + temp_repo: Path, body: str +) -> None: + with pytest.raises(SignatureRefused, match=r"nested scope"): + _rename_dry_run(temp_repo, body) + + +@pytest.mark.parametrize( + "body", + [ + " text = 'q'\n return b * a + text\n", + # A nested scope reading the new name would start reading the + # renamed parameter instead of whatever it read before. + " def inner():\n return text\n return inner() * b + a\n", + " return (lambda: times)() * b + a\n", + ], +) +def test_a_new_name_already_used_in_the_body_is_refused( + temp_repo: Path, body: str +) -> None: + with pytest.raises(SignatureRefused, match=r"already used"): + _rename_dry_run(temp_repo, body) + + +def test_a_recursive_call_is_rewritten_with_its_renamed_arguments( + temp_repo: Path, +) -> None: + # The site's arguments contain the body rename: one edit, not two + # overlapping ones. + root = _project( + temp_repo, + { + "pkg/__init__.py": "", + "pkg/util.py": ( + "def helper(a):\n return helper(a - 1) + 1 if a else 0\n" + ), + "pkg/app.py": "from pkg.util import helper\n\n\ndef run():\n return helper(2)\n", + }, + ) + store, updater = _index(root) + report = change_signature( + root, + store.fetch_all, + PROJECT, + HELPER, + ["times", "n: int"], + {"times": "a", "n": "=1"}, + reingest=updater.reingest, + ) + assert report.applied, report.message + assert report.unmapped == () + util = _read(root, "pkg/util.py") + assert "def helper(times, n: int):" in util + assert " return helper(times - 1, 1) + 1 if times else 0" in util + assert "return helper(2, 1)" in _read(root, "pkg/app.py") + assert report.verdict is not None, report.message + assert report.verdict.ok, report.message + _smoke(root, "from pkg.app import run\nassert run() == 2") + + +def test_a_call_nested_in_another_call_is_rewritten_inside_out( + temp_repo: Path, +) -> None: + root = _project( + temp_repo, + { + "pkg/__init__.py": "", + "pkg/util.py": "def helper(a):\n return a + 1\n", + "pkg/app.py": ( + "from pkg.util import helper\n\n\n" + "def run():\n return helper(helper(1))\n" + ), + }, + ) + store, updater = _index(root) + # The eval store keys a CALLS edge on its endpoints, so it holds one + # site per caller; the real store keys on (line, col) and holds both. + # Present both sites the way the real store would. + run_qn = f"{PROJECT}.pkg.app.run" + stored = store.fetch_all + + def both_sites(query: str, params: dict[str, object] | None = None): # type: ignore[no-untyped-def] + rows = stored(query, params) + out = [] + for row in rows: + if row.get(cs.KEY_QUALIFIED_NAME) != run_qn: + out.append(row) + continue + for col in (11, 18): + site = dict(row) + site[cs.KEY_COL] = col + site[cs.KEY_END_LINE] = None + site[cs.KEY_END_COL] = None + out.append(site) + return out + + report = change_signature( + root, + both_sites, + PROJECT, + HELPER, + ["a", "n: int"], + {"n": "=1"}, + reingest=updater.reingest, + ) + assert report.applied, report.message + assert report.unmapped == () + assert sum(1 for s in report.sites if s.kind == "call") == 2 + assert "return helper(helper(1, 1), 1)" in _read(root, "pkg/app.py") + # The body still adds one: helper(helper(1, 1), 1) is (1 + 1) + 1. + _smoke(root, "from pkg.app import run\nassert run() == 3") + + +# --- hierarchies --------------------------------------------------------------------- + + +SHAPES: dict[str, str] = { + "pkg/__init__.py": "", + "pkg/shapes.py": ( + "class Base:\n" + " def area(self, scale):\n" + " return 0 * scale\n\n\n" + "class Circle(Base):\n" + " def area(self, scale):\n" + " return 3 * scale\n\n\n" + "def total(shape: Base):\n" + " return shape.area(2) + Circle().area(scale=3)\n" + ), +} + + +def test_a_method_hierarchy_is_rewritten_together(temp_repo: Path) -> None: + root = _project(temp_repo, SHAPES) + store, updater = _index(root) + report = change_signature( + root, + store.fetch_all, + PROJECT, + f"{PROJECT}.pkg.shapes.Base.area", + ["scale", "unit: str"], + {"unit": "='m'"}, + reingest=updater.reingest, + ) + assert report.applied, report.message + assert set(report.hierarchy) == { + f"{PROJECT}.pkg.shapes.Base.area", + f"{PROJECT}.pkg.shapes.Circle.area", + } + shapes = _read(root, "pkg/shapes.py") + assert shapes.count("def area(self, scale, unit: str):") == 2 + assert "shape.area(2, 'm') + Circle().area(scale=3, unit='m')" in shapes + assert report.verdict is not None, report.message + assert report.verdict.ok, report.message + _smoke(root, "from pkg.shapes import total, Circle\nassert total(Circle()) == 15") + + +def test_an_override_with_different_parameters_is_refused(temp_repo: Path) -> None: + files = dict(SHAPES) + files["pkg/shapes.py"] = files["pkg/shapes.py"].replace( + " def area(self, scale):\n return 3 * scale", + " def area(self, factor):\n return 3 * factor", + ) + root = _project(temp_repo, files) + store, _updater = _index(root) + with pytest.raises(SignatureRefused, match=r"Circle\.area"): + change_signature( + root, + store.fetch_all, + PROJECT, + f"{PROJECT}.pkg.shapes.Base.area", + ["scale", "unit: str"], + {"unit": "='m'"}, + ) + assert _read(root, "pkg/shapes.py") == files["pkg/shapes.py"] + + +def test_a_method_whose_receiver_is_not_self_or_cls_is_refused( + temp_repo: Path, +) -> None: + # `this` would be remapped as a parameter and every bound call would + # lose its receiver. + files = dict(SHAPES) + files["pkg/shapes.py"] = files["pkg/shapes.py"].replace( + "def area(self, scale):", "def area(this, scale):" + ) + root = _project(temp_repo, files) + store, _updater = _index(root) + with pytest.raises(SignatureRefused, match=r"`this`.*self or cls"): + change_signature( + root, + store.fetch_all, + PROJECT, + f"{PROJECT}.pkg.shapes.Base.area", + ["scale", "unit: str"], + {"unit": "='m'"}, + ) + assert _read(root, "pkg/shapes.py") == files["pkg/shapes.py"] + + +@pytest.mark.parametrize("decorator", ["staticmethod", "builtins.staticmethod"]) +def test_a_static_method_has_no_receiver(temp_repo: Path, decorator: str) -> None: + # The indexer reads a decorator by its last name, so the operation must + # recognise the qualified spelling too. + root = _project( + temp_repo, + { + "pkg/__init__.py": "", + "pkg/shapes.py": ( + "import builtins\n\n\n" + "class K:\n" + f" @{decorator}\n" + " def helper(a):\n" + " return a\n\n\n" + "def on_class():\n" + " return K.helper(2)\n\n\n" + "def on_instance():\n" + " return K().helper(3)\n" + ), + }, + ) + store, updater = _index(root) + report = change_signature( + root, + store.fetch_all, + PROJECT, + f"{PROJECT}.pkg.shapes.K.helper", + ["a", "n: int"], + {"n": "=1"}, + reingest=updater.reingest, + ) + assert report.applied, report.message + assert report.unmapped == () + shapes = _read(root, "pkg/shapes.py") + assert " def helper(a, n: int):" in shapes + assert "return K.helper(2, 1)" in shapes + assert "return K().helper(3, 1)" in shapes + _smoke( + root, + "from pkg.shapes import on_class, on_instance\n" + "assert on_class() + on_instance() == 5", + ) + + +def test_a_static_methods_self_is_an_ordinary_parameter(temp_repo: Path) -> None: + """`@staticmethod def f(self, x)` is legal and `self` is NOT a receiver. + + The receiver was popped by NAME before staticness was decided, so + `self` was treated as the receiver AND re-added from `new_params`, + writing `def helper(self, self, a, n: int)` -- a duplicate parameter, + and a SyntaxError in the file the tool had just edited (Copilot, + #1533). Deciding staticness first makes the operation refuse and roll + back instead, leaving the source untouched. + """ + root = _project( + temp_repo, + { + "pkg/__init__.py": "", + "pkg/shapes.py": ( + "class K:\n" + " @staticmethod\n" + " def helper(self, a):\n" + " return self + a\n\n\n" + "def on_class():\n" + " return K.helper(1, 2)\n" + ), + }, + ) + store, updater = _index(root) + + report = change_signature( + root, + store.fetch_all, + PROJECT, + f"{PROJECT}.pkg.shapes.K.helper", + ["self", "a", "n: int"], + {"n": "=1"}, + reingest=updater.reingest, + ) + + # The call-site binding still counts a receiver for a static method, so + # the operation cannot map `K.helper(1, 2)` and ROLLS BACK. That is the + # correct outcome: before the reorder it wrote + # `def helper(self, self, a, n: int)` -- a duplicate parameter, and a + # SyntaxError in the file it had just edited. + assert not report.applied + assert "too_many" in report.message, report.message + shapes = _read(root, "pkg/shapes.py") + assert "def helper(self, self" not in shapes, shapes + assert " def helper(self, a):" in shapes, shapes + + +def test_a_bare_generator_argument_is_unmapped_not_rewritten( + temp_repo: Path, +) -> None: + """`helper(x for x in xs)` passes a GENERATOR, not a list of values. + + tree-sitter gives the call a `generator_expression` as its arguments + node rather than an `argument_list`, so binding it positionally would + rewrite a generator as if it were an ordinary value and change what + the call means. It must be listed unmapped instead (Copilot, #1533). + """ + root = _project( + temp_repo, + { + "pkg/__init__.py": "", + "pkg/app.py": ( + "def helper(a):\n" + " return sum(a)\n\n\n" + "def run(xs):\n" + " return helper(x for x in xs)\n" + ), + }, + ) + store, updater = _index(root) + + report = change_signature( + root, + store.fetch_all, + PROJECT, + f"{PROJECT}.pkg.app.helper", + ["a", "n: int"], + {"n": "=1"}, + reingest=updater.reingest, + ) + + (skipped,) = report.unmapped + assert skipped.path == "pkg/app.py" + # The REASON matters, not just the refusal: without the argument-list + # check the call is still skipped, but by an arity miscount that counts + # the generator's own children as two positional arguments. That is the + # right answer for the wrong reason, and it would stop being right for + # a one-parameter generator call. + assert cs.SIGNATURE_SITE_UNREADABLE.split("{")[0] in skipped.reason, skipped.reason + # The generator survives verbatim: it was never bound as a value. + assert "helper(x for x in xs)" in _read(root, "pkg/app.py") + + +def test_a_default_reading_a_renamed_parameter_refuses(temp_repo: Path) -> None: + """`def f(a, b=a)` renamed a->x would write `def f(x, b=a)`. + + A default is evaluated at DEFINITION time in the enclosing scope, so + that file raises NameError the moment it is imported -- and it PARSES, + so the syntax postcondition cannot catch it. `_body_references` walks + the body only and never saw the reference (Copilot, #1533). + """ + root = _project( + temp_repo, + { + "pkg/__init__.py": "", + "pkg/app.py": "def f(a, b=a):\n return b\n", + }, + ) + store, updater = _index(root) + + with pytest.raises(SignatureRefused, match="evaluated at definition time"): + change_signature( + root, + store.fetch_all, + PROJECT, + f"{PROJECT}.pkg.app.f", + ["x", "b=a"], + {"x": "a"}, + reingest=updater.reingest, + ) + + assert "def f(a, b=a):" in _read(root, "pkg/app.py") + + +def test_a_default_not_reading_the_renamed_parameter_still_renames( + temp_repo: Path, +) -> None: + """The control: a default that does NOT read the renamed name is fine, + so the refusal cannot be satisfied by refusing every default.""" + root = _project( + temp_repo, + { + "pkg/__init__.py": "", + "pkg/app.py": "def f(a, b=1):\n return a + b\n", + }, + ) + store, updater = _index(root) + + report = change_signature( + root, + store.fetch_all, + PROJECT, + f"{PROJECT}.pkg.app.f", + ["x", "b=1"], + {"x": "a"}, + reingest=updater.reingest, + ) + + assert report.applied, report.message + assert "def f(x, b=1):" in _read(root, "pkg/app.py") + + +def test_an_explicit_spelling_re_annotates_every_override( + temp_repo: Path, +) -> None: + """Spelling a kept parameter out again re-annotates it everywhere. + + `carried` was inferred by comparing each new spec with the primary's, + and a bare name resolves TO the primary's spec -- so an explicit + `a: int` that matched the primary compared equal, counted as carried, + and left an override declaring `a: str` untouched (Copilot, #1533). + """ + root = _project( + temp_repo, + { + "pkg/__init__.py": "", + "pkg/app.py": ( + "class Base:\n def f(self, a: int) -> int:\n return a\n\n\n" + "class Sub(Base):\n def f(self, a: str) -> int:\n" + " return len(a)\n" + ), + }, + ) + store, updater = _index(root) + + report = change_signature( + root, + store.fetch_all, + PROJECT, + f"{PROJECT}.pkg.app.Base.f", + ["a: int", "n: int"], + {"n": "=1"}, + reingest=updater.reingest, + ) + + assert report.applied, report.message + app = _read(root, "pkg/app.py") + assert "def f(self, a: int, n: int) -> int:" in app, app + assert "a: str" not in app, app + + +def test_a_bare_name_keeps_each_overrides_own_spelling(temp_repo: Path) -> None: + """The control: a BARE name carries each override's own annotation, so + the fix above cannot be satisfied by re-annotating unconditionally.""" + root = _project( + temp_repo, + { + "pkg/__init__.py": "", + "pkg/app.py": ( + "class Base:\n def f(self, a: int) -> int:\n return a\n\n\n" + "class Sub(Base):\n def f(self, a: str) -> int:\n" + " return len(a)\n" + ), + }, + ) + store, updater = _index(root) + + report = change_signature( + root, + store.fetch_all, + PROJECT, + f"{PROJECT}.pkg.app.Base.f", + ["a", "n: int"], + {"n": "=1"}, + reingest=updater.reingest, + ) + + assert report.applied, report.message + app = _read(root, "pkg/app.py") + assert "def f(self, a: int, n: int) -> int:" in app, app + assert "def f(self, a: str, n: int) -> int:" in app, app + + +def test_dropping_a_parameter_the_body_reads_refuses(temp_repo: Path) -> None: + """`def f(a): return a` emptied writes `def f(): return a`. + + That PARSES, so the syntax postcondition passes, and the arity + contract only checks call sites -- the NameError surfaces when the + function is finally called (Copilot, #1533). + """ + root = _project( + temp_repo, + { + "pkg/__init__.py": "", + "pkg/app.py": "def f(a):\n return a\n\n\ndef call():\n return f(1)\n", + }, + ) + store, updater = _index(root) + + with pytest.raises(SignatureRefused, match="still reads"): + change_signature( + root, + store.fetch_all, + PROJECT, + f"{PROJECT}.pkg.app.f", + [], + None, + reingest=updater.reingest, + ) + + assert "def f(a):" in _read(root, "pkg/app.py") + + +def test_dropping_a_parameter_the_body_ignores_still_applies( + temp_repo: Path, +) -> None: + """The control: a parameter the body never reads drops cleanly, so the + refusal cannot be satisfied by refusing every removal.""" + root = _project( + temp_repo, + { + "pkg/__init__.py": "", + "pkg/app.py": "def f(a):\n return 1\n\n\ndef call():\n return f(2)\n", + }, + ) + store, updater = _index(root) + + report = change_signature( + root, + store.fetch_all, + PROJECT, + f"{PROJECT}.pkg.app.f", + [], + None, + reingest=updater.reingest, + ) + + assert report.applied, report.message + app = _read(root, "pkg/app.py") + assert "def f():" in app, app + assert "return f()" in app, app + + +@pytest.mark.parametrize( + "body", + [ + "def f(a):\n return obj.a\n", + "def f(a):\n return helper(a=1)\n", + ], + ids=["attribute-name", "keyword-label"], +) +def test_a_same_named_attribute_or_keyword_is_not_a_read( + temp_repo: Path, body: str +) -> None: + """`obj.a` and `helper(a=1)` spell `a` without reading the parameter. + + The drop guard matched any identifier, so both refused a removal that + is perfectly legal -- the guard has to use the same `_is_a_reference` + test the body rewrite does (CodeRabbit, #1533). + """ + root = _project( + temp_repo, + { + "pkg/__init__.py": "", + "pkg/app.py": body + "\n\ndef call():\n return f(1)\n", + }, + ) + store, updater = _index(root) + + report = change_signature( + root, + store.fetch_all, + PROJECT, + f"{PROJECT}.pkg.app.f", + [], + None, + reingest=updater.reingest, + ) + + assert report.applied, report.message + assert "def f():" in _read(root, "pkg/app.py") + + +# --- transaction and contract --------------------------------------------------------- + + +def test_dry_run_reports_the_diff_and_writes_nothing( + repo: tuple[Path, _StatefulIngestor, GraphUpdater], +) -> None: + root, store, _updater = repo + report = change_signature( + root, + store.fetch_all, + PROJECT, + HELPER, + ["a", "n: int", "b"], + {"n": "=1"}, + dry_run=True, + ) + assert not report.applied + assert "--- a/pkg/util.py" in report.diff + assert "+def helper(a: int, n: int, b: str = 'x') -> str:" in report.diff + for rel, text in FIXTURE.items(): + assert _read(root, rel) == text + assert load_history(root) == [] + + +def test_the_applied_change_is_recorded_for_undo( + repo: tuple[Path, _StatefulIngestor, GraphUpdater], +) -> None: + root, store, _updater = repo + report = change_signature( + root, store.fetch_all, PROJECT, HELPER, ["a", "n: int", "b"], {"n": "=1"} + ) + assert report.applied, report.message + assert report.verdict is None # no re-ingest, so no contract + (entry,) = load_history(root) + assert entry[cs.EDIT_KEY_ID] == report.transaction_id + + +def test_a_contract_failure_undoes_the_change( + repo: tuple[Path, _StatefulIngestor, GraphUpdater], +) -> None: + root, store, updater = repo + # A caller the graph never saw: the operation cannot rewrite it, and once + # the re-ingest brings it in, its site passes too few arguments for the + # new signature without being listed as unmapped. The contract catches + # it after the fact and the transaction is undone. + _write( + root, + "pkg/late.py", + "from pkg.util import helper\n\n\ndef late():\n return helper(2)\n", + ) + before = {rel: _read(root, rel) for rel in FIXTURE} + report = change_signature( + root, + store.fetch_all, + PROJECT, + HELPER, + ["a", "n: int", "b"], + {"n": "=1"}, + reingest=lambda paths: updater.reingest([*paths, "pkg/late.py"]), + ) + assert not report.applied + assert report.verdict is not None + assert not report.verdict.ok + assert "pkg/late.py:5" in report.message + for rel, text in before.items(): + assert _read(root, rel) == text + assert load_history(root) == [] + + +# --- MCP tool ---------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_mcp_change_signature_tool_runs_under_the_lock_and_reports( + repo: tuple[Path, _StatefulIngestor, GraphUpdater], +) -> None: + from unittest.mock import MagicMock + + from codebase_rag.mcp.tools import MCPToolsRegistry + + root, store, _updater = repo + ingestor = MagicMock() + ingestor.fetch_all = store.fetch_all + ingestor.list_projects.return_value = [PROJECT] + registry = MCPToolsRegistry( + project_root=str(root), ingestor=ingestor, cypher_gen=MagicMock() + ) + schema = next( + s + for s in registry.get_tool_schemas() + if s.name == cs.MCPToolName.CHANGE_SIGNATURE + ) + assert set(schema.inputSchema["required"]) == { + cs.MCPParamName.QUALIFIED_NAME, + cs.MCPParamName.NEW_PARAMS, + } + entry = registry.get_tool_handler(cs.MCPToolName.CHANGE_SIGNATURE) + assert entry is not None + payload = await entry[0]( + qualified_name=HELPER, + new_params=["a", "n: int", "b"], + mapping={"n": "=1"}, + dry_run=True, + project=PROJECT, + ) + assert isinstance(payload, dict) + assert payload["applied"] is False + assert payload[cs.KEY_SITES] + assert payload["unmapped"] == [] + assert "helper(2, 1)" in payload["diff"] + assert _read(root, "pkg/app.py") == FIXTURE["pkg/app.py"] + + +@pytest.mark.asyncio +async def test_mcp_change_signature_refusal_is_a_payload_not_an_exception( + repo: tuple[Path, _StatefulIngestor, GraphUpdater], +) -> None: + from unittest.mock import MagicMock + + from codebase_rag.mcp.tools import MCPToolsRegistry + + root, store, _updater = repo + ingestor = MagicMock() + ingestor.fetch_all = store.fetch_all + ingestor.list_projects.return_value = [PROJECT] + registry = MCPToolsRegistry( + project_root=str(root), ingestor=ingestor, cypher_gen=MagicMock() + ) + payload = await registry.change_signature( + qualified_name=HELPER, + new_params=["a", "n: int", "b"], + mapping={"n": "nothing"}, + project=PROJECT, + ) + assert isinstance(payload, dict) + assert cs.DICT_KEY_ERROR in payload + assert "nothing" in payload[cs.DICT_KEY_ERROR] diff --git a/codebase_rag/tests/test_edit_contract.py b/codebase_rag/tests/test_edit_contract.py index 0addf69d7c..12f974fee2 100644 --- a/codebase_rag/tests/test_edit_contract.py +++ b/codebase_rag/tests/test_edit_contract.py @@ -68,12 +68,12 @@ def _dangling(path: str = "pkg/app.py", line: int = 5) -> dict[str, object]: } -def _site(path: str, line: int, verdict: str) -> dict[str, object]: +def _site(path: str, line: int, verdict: str, col: int = 11) -> dict[str, object]: return { "caller": "p.pkg.app.run", "path": path, "line": line, - "col": 11, + "col": col, "arg_count": 2, "kwarg_names": [], "declared_count": 1, @@ -383,6 +383,22 @@ def test_rename_refuses_silently_rewritten_heuristic_sites() -> None: assert verify(allowed, delta, rewritten=rewritten).ok +def test_change_signature_refuses_silently_rewritten_heuristic_sites() -> None: + # The same leave a rename carries: without it, `allow_heuristic` on a + # signature change rewrote the guessed site and the contract then undid + # the whole change for having done what it was told. + delta = _delta( + symbols={"added": [], "removed": [], "renamed": [], "changed": ["p.h"]} + ) + rewritten = [("pkg/app.py:5", "exact"), ("pkg/other.py:9", "heuristic")] + strict = verify(change_signature_expectation([]), delta, rewritten=rewritten) + assert strict.failures == ( + cs.CONTRACT_HEURISTIC_REWRITTEN.format(sites="pkg/other.py:9"), + ) + allowed = change_signature_expectation([], heuristic_allowed=True) + assert verify(allowed, delta, rewritten=rewritten).ok + + def test_change_signature_requires_every_site_mapped_or_listed() -> None: change = { "qualified_name": "p.pkg.util.helper", @@ -408,23 +424,61 @@ def test_change_signature_requires_every_site_mapped_or_listed() -> None: strict = verify(change_signature_expectation([]), delta) assert strict.failures == ( cs.CONTRACT_SITES_UNMAPPED.format( - sites="pkg/app.py:9 (possibly_missing), pkg/cli.py:3 (too_many)" + sites="pkg/app.py:9:11 (possibly_missing), pkg/cli.py:3:11 (too_many)" ), ) listed = verify( - change_signature_expectation(["pkg/app.py:9", "pkg/cli.py:3"]), delta + change_signature_expectation(["pkg/app.py:9:11", "pkg/cli.py:3:11"]), delta ) assert listed.ok # A site the operation rewrote is mapped by construction: a # `possibly_missing` there relies on a default the mapping supplied. rewritten = verify( - change_signature_expectation(["pkg/cli.py:3"]), + change_signature_expectation(["pkg/cli.py:3:11"]), delta, - rewritten=[("pkg/app.py:9", "exact")], + rewritten=[("pkg/app.py:9:11", "exact")], ) assert rewritten.ok +def test_change_signature_tells_two_calls_on_one_line_apart() -> None: + # `helper(1) + helper(2)`: rewriting or listing one call must not cover + # the other, so the site's identity carries its column. + change = { + "qualified_name": "p.pkg.util.helper", + "path": "pkg/util.py", + "before": ["a"], + "after": ["a", "b"], + "sites": [ + _site("pkg/app.py", 9, cs.DELTA_ARITY_POSSIBLY_MISSING, col=4), + _site("pkg/app.py", 9, cs.DELTA_ARITY_POSSIBLY_MISSING, col=20), + ], + } + delta = _delta( + symbols={ + "added": [], + "removed": [], + "renamed": [], + "changed": ["p.pkg.util.helper"], + }, + signature_changes=[change], + ) + one_rewritten = verify( + change_signature_expectation([]), + delta, + rewritten=[("pkg/app.py:9:4", "exact")], + ) + assert one_rewritten.failures == ( + cs.CONTRACT_SITES_UNMAPPED.format(sites="pkg/app.py:9:20 (possibly_missing)"), + ) + one_listed = verify( + change_signature_expectation(["pkg/app.py:9:4"]), + delta, + rewritten=[("pkg/app.py:9:20", "exact")], + ) + assert one_listed.ok + + def test_move_requires_no_new_cycle_and_updated_importers() -> None: expectation = move_expectation("p.pkg.util.helper", "p.pkg.core.helper") moved = [ diff --git a/codebase_rag/tools/tool_descriptions.py b/codebase_rag/tools/tool_descriptions.py index 91b7b948a9..f9afade85a 100644 --- a/codebase_rag/tools/tool_descriptions.py +++ b/codebase_rag/tools/tool_descriptions.py @@ -283,6 +283,31 @@ class AgenticToolName(StrEnum): "Rewrite through heuristic, overload and trace-only sites too (default false)." ) MCP_PARAM_RENAME_DRY_RUN = "Plan only: report the sites and diff, write nothing." +MCP_CHANGE_SIGNATURE = ( + "Change a Python function's or method's parameter list and rewrite every " + "call site the graph knows, on the definition and on every override in " + "both directions. `new_params` lists the new parameters in order (`n: int`, " + "`limit: int = 10`, or a bare old name to carry it over unchanged). " + "`mapping` says where each new parameter's value comes from at a call site: " + "an old parameter's name, its 0-based index, or `=literal` to insert that " + "text at every site; a new parameter absent from the mapping takes the old " + "parameter of the same name. A site the mapping cannot complete (a missing " + "required value, a splat, a surplus argument, an undeclared keyword) or " + "that the graph resolved heuristically is left as written and listed as " + "`unmapped`; pass `allow_heuristic` to rewrite guessed sites too. Every " + "rewritten file must still parse, and with an indexed project the change " + "is undone when its postcondition fails. Set `dry_run` to see the plan and " + "diff without touching the tree. Applied edits are recorded for `cgr edits " + "undo`." +) +MCP_PARAM_NEW_PARAMS = ( + "The new parameter list in order, one entry per parameter (`name`, " + "`name: type`, `name = default`, `name: type = default`)." +) +MCP_PARAM_MAPPING = ( + "Optional. New parameter name to its source: an old parameter name, a " + "0-based old index, or `=literal`." +) MCP_QUERY_CODE_GRAPH = ( "Prefer the deterministic tools (resolve, definition, callers, callees, " @@ -469,6 +494,7 @@ class AgenticToolName(StrEnum): MCPToolName.ANNOTATE: MCP_ANNOTATE, MCPToolName.GLOSSES: MCP_GLOSSES, MCPToolName.RENAME: MCP_RENAME, + MCPToolName.CHANGE_SIGNATURE: MCP_CHANGE_SIGNATURE, MCPToolName.QUERY_CODE_GRAPH: MCP_QUERY_CODE_GRAPH, MCPToolName.GET_CODE_SNIPPET: MCP_GET_CODE_SNIPPET, MCPToolName.SURGICAL_REPLACE_CODE: MCP_SURGICAL_REPLACE_CODE, diff --git a/docs/architecture/change-signature.md b/docs/architecture/change-signature.md new file mode 100644 index 0000000000..900ac2988e --- /dev/null +++ b/docs/architecture/change-signature.md @@ -0,0 +1,162 @@ +--- +description: "How cgr changes a Python definition's parameter list through the graph: the definition, its overrides and every known call site are rewritten in one transaction, and sites the mapping cannot complete are listed rather than guessed." +--- + +# Change Signature Operation + +`change_signature` is the second graph-native edit operation (issue #1533). +Given a qualified name, the new parameter list and a mapping from new +parameters to old values, it rewrites the definition (and every override in +both directions) and every call site the graph knows, atomically, or +explains what it left alone and why. + +```bash +# add a required parameter, passing 1 at every existing site +cgr change-signature myproj.pkg.util.helper a 'n: int' b --map n==1 --dry-run +# reorder +cgr change-signature myproj.pkg.util.helper b a +# rename by position: keyword callers follow the new names +cgr change-signature myproj.pkg.util.helper 'times: int' 'text: str' --map times=0 --map text=1 +``` + +The MCP tool of the same name takes `qualified_name`, `new_params` (a list +of strings) and the optional `mapping` (an object), `allow_heuristic`, +`dry_run` and `project` fields, and returns the same report as JSON. + +Python only for now. A definition in another language refuses with a +message naming the language, and the follow-up is issue #1908. + +## The new parameter list + +Each entry of `new_params` is one parameter as it would be written in a +header: `n`, `n: int`, `n = 0` or `n: int = 0`. A bare name that matches an +old parameter carries that parameter over with its existing annotation and +default, so `["b", "a"]` on `def helper(a: int, b: str = 'x')` produces +`def helper(b: str = 'x', a: int)`, which is then refused because a required +parameter cannot follow a defaulted one. Spelling a kept parameter out +again re-annotates it: `["a", "b: int = 0"]` writes `b: int = 0` into the +header, and a literal mapped to it is checked against the new annotation. + +The header must consist of plain positional-or-keyword parameters. A +definition with `*args`, `**kwargs`, a positional-only `/` or a keyword-only +`*` marker refuses: those change how call sites bind and the mapping below +cannot express them. A method's `self` or `cls` is kept in place and never +part of the list; a `@staticmethod` has no receiver and every parameter is +listed. A method whose first parameter is named anything else refuses, +since the receiver could not be told from the parameters and every bound +call would lose it. + +Every override in the hierarchy (`OVERRIDES` edges in both directions) must +declare the same parameter names, or the operation refuses naming the +member that differs: rewriting one and not the other would break the +substitution the hierarchy exists for. + +## The mapping + +The mapping says, for each new parameter, where its value comes from at a +call site: + +| Source | Meaning | +|---------------|----------------------------------------------------------------------| +| `old_name` | The value the site passed for that old parameter. | +| `0`, `1`, ... | The same, by the old parameter's position. | +| `=literal` | That text, inserted at every site. The definition does NOT gain a default: the new parameter is as required or defaulted as its entry says. | + +A new parameter absent from the mapping takes the old parameter of the same +name, if there is one. On the CLI the entries are `--map NEW=SOURCE`, so +`--map n==1` maps `n` to the literal `1`. + +The mapping is refused, and nothing is written, when it names a parameter +that is not in the new list, feeds one from an old name or index that does +not exist, feeds two new parameters from the same old one, or gives a +literal that cannot be a value of the parameter's declared type: `=None` +fits `int | None` and `Optional[int]`, `='no'` does not fit `int`, and a +non-literal such as `=LIMIT` or an annotation the check does not read is +not checked. + +## Renamed parameters + +An old parameter fed to a new one of a different name (`times` from `a`) +is renamed, and the body follows: every bare reference to the old name in +the function's body becomes the new name, in the definition and in every +override. An attribute (`obj.a`) and a keyword argument's name (`f(a=a)` +becomes `f(a=times)`) share the spelling without being the parameter and +are left alone. A comprehension that reads the parameter follows the +rename too, since it binds only its own `for` targets. + +The rename refuses, with nothing written, when the old name is used inside +a nested function, lambda or class, or re-bound by a `global`, `nonlocal` +or import statement, because the body walk cannot tell those uses from the +parameter's; and when the new name is already read in the body, or inside +a nested scope in it, because the parameter would shadow it. Swapping two +parameters (`b` from `a`, `a` from `b`) is a rename in each direction and +is allowed. Docstrings are prose and are not rewritten. + +A call to the function inside its own body, or a call nested in another +call's arguments, is rewritten inside out: the inner rewrite (a renamed +argument, an inner site's new argument list) is folded into the enclosing +value before the enclosing site is rendered, so `helper(helper(1))` becomes +`helper(helper(1, 1), 1)` as one edit per site rather than two that overlap. + +## What gets rewritten + +| Site kind | Source | +|--------------|------------------------------------------------------------------------| +| `definition` | The `parameters` node of the definition and of every override, replaced with the new list. | +| `call` | `CALLS` edges into the definition or an override, using the per-site `line`/`col`/`end_line`/`end_col` recorded at ingest (see [graph schema](graph-schema.md#edge-site-properties)); the argument list is re-rendered. | + +At a call site the arguments are bound to the old parameters as Python +would bind them, then re-rendered in the new order. Values keep positional +form until one is spelled by keyword or a defaulted parameter is left out; +from there every later value goes by keyword, because it can no longer sit +in its slot. Values that were keywords keep their original relative order +under their new names. So with `helper(a, b='x')` becoming +`helper(a, b='x', n=0)` and `n` mapped to `=1`: + +| Before | After | +|----------------------|---------------------------| +| `helper(2)` | `helper(2, n=1)` | +| `helper(2, b='y')` | `helper(2, b='y', n=1)` | +| `helper(3, 'z')` | `helper(3, 'z', 1)` | + +A site whose rendered text equals the original is counted but not edited. +Every edit is a span-preserving [patcher](patchers.md) edit; a multi-line +argument list is rendered on one line. + +## Unmapped sites + +A site is left exactly as written, and listed in `unmapped` with its owner, +path, line, column and reason, when: + +- it passes no value for a new parameter without a default and the mapping + supplies none; +- it passes more positional arguments than the definition declares, a + keyword the definition does not declare, or the same parameter twice; +- it uses `*args`, `**kwargs`, or a bare generator as its arguments; +- the graph bound it by `heuristic`, `overload` or `dynamic` resolution and + `allow_heuristic` is false; +- it carries no location, or its file cannot be read. + +Unlike `rename`, an unmapped site does not refuse the operation: the +definition still changes, and the list is what the caller (or the +[postcondition contract](postcondition-contract.md)) checks. The contract +requires every call site of the changed signature to read as mapped after +the re-ingest or to be in that list, so a caller the graph never saw is +caught after the fact and the transaction is undone. + +## Transaction and contract + +The definition and the sites are patched into one +[edit transaction](edit-transactions.md). Every patched file must still +parse, or nothing is written and the report says which files would break. +`dry_run` stages, reports the diff and rolls back. An applied change is +recorded for `cgr edits undo`. + +With a re-ingest available (always on the CLI and the MCP server, when the +project is indexed from this tree) the applied change is measured through +the [structural delta](structural-delta.md) and checked against the +contract: symbol and caller counts unchanged, no dangling callers, every +site mapped or listed, no guessed site rewritten without `allow_heuristic`. +A failure undoes the transaction, re-ingests the restored files and reports +the reasons; `verdict.affected_tests` lists the tests reaching the changed +symbols either way. diff --git a/docs/guide/mcp-server.md b/docs/guide/mcp-server.md index 35ae4a0099..d3594463ab 100644 --- a/docs/guide/mcp-server.md +++ b/docs/guide/mcp-server.md @@ -75,6 +75,7 @@ cgr daemon up | `annotate` | Attach a durable note (a Gloss) to one definition in the graph, never to the source file. `target` names the definition the way `resolve` does; a name matching several definitions is refused with the candidates, so pass a qualified name to disambiguate. `kind` types the claim (invariant, mirrors, platform-conditional, safety-precondition); `mentions` lists other definitions the note refers to, each becoming a MENTIONS edge. Write the WHY that is not in the code ("safe because validate() runs first"), not a restatement of what the code does. Returns the stored gloss, or an error and nothing written. | | `glosses` | Notes (Glosses) about one definition: `annotating` are filed on it, `mentioning` are filed on other definitions and refer to it. `target` is resolved as `annotate` resolves it. Each note's `anchor_state` is graded after each sync for function and method subjects: EXACT when the definition's current content hash equals the one recorded when the note was written, STALE when it differs (and EXACT again if the code is reverted). A note on a class or module, or one written before hashes were recorded, keeps the state it has. Deterministic: fixed graph queries, no LLM, same graph gives the same JSON. Use this instead of query_code_graph whenever you know the exact name or location. | | `rename` | Rename a function, method, class or other definition everywhere the graph knows it is referenced: the definition, every call, reference and instantiation site, import statements (aliases kept), overrides in both directions, and Python `__all__` entries. Refuses when any site was resolved heuristically, by overload fan-out, or only by a trace, unless `allow_heuristic` is true. A class referenced by an inheritance or type annotation edge that carries no rewrite location refuses regardless of `allow_heuristic`. Every rewritten file must still parse; otherwise nothing is written. Set `dry_run` to see the plan and diff without touching the tree. Applied edits are recorded for `cgr edits undo`, and Markdown mentions of the old name are listed for a human to review. | +| `change_signature` | Change a Python function's or method's parameter list and rewrite every call site the graph knows, on the definition and on every override in both directions. `new_params` lists the new parameters in order (`n: int`, `limit: int = 10`, or a bare old name to carry it over unchanged). `mapping` says where each new parameter's value comes from at a call site: an old parameter's name, its 0-based index, or `=literal` to insert that text at every site; a new parameter absent from the mapping takes the old parameter of the same name. A site the mapping cannot complete (a missing required value, a splat, a surplus argument, an undeclared keyword) or that the graph resolved heuristically is left as written and listed as `unmapped`; pass `allow_heuristic` to rewrite guessed sites too. Every rewritten file must still parse, and with an indexed project the change is undone when its postcondition fails. Set `dry_run` to see the plan and diff without touching the tree. Applied edits are recorded for `cgr edits undo`. | | `query_code_graph` | Prefer the deterministic tools (resolve, definition, callers, callees, implementors, overrides, importers, tests_reaching) when you know the exact name or location: they run fixed queries with no LLM. Use this for open-ended questions. Query the codebase knowledge graph using natural language. Ask questions like 'What functions call UserService.create_user?' or 'Show me all classes that implement the Repository interface'. Pass `project` to restrict results to one indexed project; use list_projects for the available names. Omit it to search them all. The scope is enforced on the results before they are capped, so it holds regardless of the query generated and the result limit is spent on rows from the requested project. | | `get_code_snippet` | Retrieve source code for a function, class, or method by its qualified name. Returns the source code, file path, line numbers, and docstring. | | `surgical_replace_code` | Surgically replace an exact code block in a file using diff-match-patch. Only modifies the exact target block, leaving the rest unchanged. After a successful write on an indexed project, the touched files are re-ingested and a structural delta is appended to the result as JSON: symbols added, removed and renamed, callers left dangling, call sites passing too many arguments, signature changes with a verdict per call site, new duplicates of existing functions, new import cycles, and the tests reaching the edited symbols. A project that is not indexed appends nothing, and a failed analysis appends an error note instead of a delta. Read it before the next edit. | diff --git a/mkdocs.yml b/mkdocs.yml index 3f2a7052cf..7ebd680da0 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -110,6 +110,7 @@ nav: - Span-Preserving Patchers: architecture/patchers.md - Edit Transactions: architecture/edit-transactions.md - Rename Operation: architecture/rename.md + - Change Signature Operation: architecture/change-signature.md - Postcondition Contract: architecture/postcondition-contract.md - Advanced: - Adding Languages: advanced/adding-languages.md