From 2a04ce4acd3a755504372f20d367595dee1a77da Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 16 Sep 2026 14:06:55 +0000 Subject: [PATCH] feat: suggest reviewable migration adapters from diffs (#62) Add suggest_adapter library helper and suggest-adapter CLI that drafts ToolAlias / ArgumentMap / EnumMap JSON from rename and schema signals. Drafts always set auto_apply=false with confidence notes; never applied by CI. Co-authored-by: Abhinaysai Kamineni --- docs/adapters.md | 41 ++++ src/tool_semantics/cli.py | 71 +++++++ src/tool_semantics/suggest.py | 367 ++++++++++++++++++++++++++++++++++ tests/test_suggest.py | 109 ++++++++++ 4 files changed, 588 insertions(+) create mode 100644 src/tool_semantics/suggest.py create mode 100644 tests/test_suggest.py diff --git a/docs/adapters.md b/docs/adapters.md index 30e1ef7..8856e65 100644 --- a/docs/adapters.md +++ b/docs/adapters.md @@ -33,6 +33,47 @@ payload = proxy.route_result(tool, {"items": []}) - Adapters are the **runtime** counterpart: they keep old call shapes working. - Prefer documenting adapters next to intentional renames in release notes. +## Suggesting adapter drafts (#62) + +Generate a **reviewable** draft from two snapshots. Suggestions are never +auto-applied by CI — humans must review and opt in before shipping. + +```bash +tool-semantics capture examples/github_server_v1.json -o .tool-semantics/v1.json +tool-semantics capture examples/github_server_v2.json -o .tool-semantics/v2.json +tool-semantics suggest-adapter .tool-semantics/v1.json .tool-semantics/v2.json \ + -o .tool-semantics/adapter.suggested.json +``` + +```python +from tool_semantics.scanner import capture_manifest +from tool_semantics.suggest import suggest_adapter +from pathlib import Path + +baseline = capture_manifest(Path("examples/github_server_v1.json")) +candidate = capture_manifest(Path("examples/github_server_v2.json")) +draft = suggest_adapter(baseline, candidate) +assert draft.auto_apply is False +print(draft.adapter.aliases, draft.notes) +``` + +### What gets suggested + +| Signal | Action | Confidence | +| --- | --- | --- | +| `tool.renamed` in the report | `ToolAlias` | high | +| Remaining remove+add above suggestion threshold | `ToolAlias` | medium/low | +| Removed+added parameters with name similarity | `ArgumentMap.rename` | high/medium/low | +| Enums with equal cardinality + unique token matches | `EnumMap` | medium | +| Required parameter adds / tied enums / cardinality mismatch | skipped | skipped | + +### Confidence limits + +- Heuristic renames can be wrong when two unrelated tools look similar. +- Enum remaps are skipped when values cannot be paired unambiguously. +- New required parameters never get invented defaults. +- Treat every draft as a starting point for review, not production config. + ## MCP compatibility proxy `CompatibilityProxy` is an in-process shim. A network MCP proxy that speaks diff --git a/src/tool_semantics/cli.py b/src/tool_semantics/cli.py index 51a515c..2b42a4f 100644 --- a/src/tool_semantics/cli.py +++ b/src/tool_semantics/cli.py @@ -38,6 +38,7 @@ ) from tool_semantics.runner import OpenAICompatibleRunner, RunnerConfig from tool_semantics.scanner import ManifestError, capture_manifest, read_snapshot, write_snapshot +from tool_semantics.suggest import adapter_suggestion_to_json, suggest_adapter app = typer.Typer( no_args_is_help=True, @@ -702,3 +703,73 @@ def compare( markdown_output.write_text(render_markdown(report), encoding="utf-8") if fails_policy: raise typer.Exit(code=1) + + +@app.command("suggest-adapter") +def suggest_adapter_cmd( + baseline: Annotated[ + Path, + typer.Argument(dir_okay=False, help="Baseline snapshot JSON from `capture`."), + ], + candidate: Annotated[ + Path, + typer.Argument(dir_okay=False, help="Candidate snapshot JSON from `capture`."), + ], + output: Annotated[ + Path | None, + typer.Option( + "--output", + "-o", + help="Write reviewable adapter suggestion JSON (never auto-applied).", + ), + ] = None, + verbose: Annotated[ + bool, + typer.Option("--verbose", "-v", help="Log suggestion steps to stderr."), + ] = False, +) -> None: + """Suggest a MigrationAdapter draft from rename / schema diffs (#62). + + Output is reviewable JSON only — CI must not apply it without an explicit + human opt-in step outside this command. + """ + _require_snapshot_file(baseline, "Baseline") + _require_snapshot_file(candidate, "Candidate") + try: + baseline_snap = read_snapshot(baseline) + candidate_snap = read_snapshot(candidate) + report = compare_snapshots(baseline_snap, candidate_snap) + suggestion = suggest_adapter(baseline_snap, candidate_snap, report=report) + except (ManifestError, FileNotFoundError, ValueError) as exc: + console.print(f"[red]Adapter suggestion failed:[/red] {exc}") + raise typer.Exit(code=2) from exc + + _log_verbose( + verbose, + f"Aliases={len(suggestion.adapter.aliases)} " + f"arg_maps={len(suggestion.adapter.arguments)} " + f"enums={len(suggestion.adapter.enums)} notes={len(suggestion.notes)}", + ) + console.print( + f"Suggested adapter draft: " + f"{len(suggestion.adapter.aliases)} alias(es), " + f"{len(suggestion.adapter.arguments)} argument map(s), " + f"{len(suggestion.adapter.enums)} enum map(s). " + f"[yellow]auto_apply={suggestion.auto_apply}[/yellow] — review before use." + ) + if suggestion.notes: + table = Table(title="Suggestion notes") + table.add_column("Confidence") + table.add_column("Subject") + table.add_column("Message") + for note in suggestion.notes: + table.add_row(note.confidence.value, note.subject, note.message) + console.print(table) + + payload = adapter_suggestion_to_json(suggestion) + if output is not None: + output.parent.mkdir(parents=True, exist_ok=True) + output.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") + console.print(f"Wrote reviewable draft to {output}") + else: + console.print_json(data=payload) diff --git a/src/tool_semantics/suggest.py b/src/tool_semantics/suggest.py new file mode 100644 index 0000000..3e69b11 --- /dev/null +++ b/src/tool_semantics/suggest.py @@ -0,0 +1,367 @@ +"""Suggest reviewable MigrationAdapter drafts from compatibility diffs (#62). + +Suggestions are never applied automatically — emit JSON/YAML for human review. +""" + +from __future__ import annotations + +from difflib import SequenceMatcher +from enum import StrEnum +from typing import Any + +from pydantic import BaseModel, Field + +from tool_semantics.adapters import ( + ArgumentMap, + EnumMap, + MigrationAdapter, + ToolAlias, +) +from tool_semantics.diff import ( + CompatibilityReport, + _detect_renames, + _jaccard, + _token_set, + _tool_similarity, + compare_snapshots, +) +from tool_semantics.models import InterfaceSnapshot, ToolContract, ToolParameter + + +class SuggestionConfidence(StrEnum): + HIGH = "high" + MEDIUM = "medium" + LOW = "low" + SKIPPED = "skipped" + + +class SuggestionNote(BaseModel): + """Why a mapping was included or skipped.""" + + subject: str + confidence: SuggestionConfidence + message: str + + +class AdapterSuggestion(BaseModel): + """Reviewable adapter draft plus confidence notes (never auto-applied).""" + + adapter: MigrationAdapter = Field(default_factory=MigrationAdapter) + notes: list[SuggestionNote] = Field(default_factory=list) + baseline: str = "" + candidate: str = "" + auto_apply: bool = False # Always false; CI must not apply without explicit opt-in. + + +def _param_similarity(left: str, right: str) -> float: + token = _jaccard(_token_set(left.replace("_", " ")), _token_set(right.replace("_", " "))) + seq = SequenceMatcher(None, left.lower(), right.lower()).ratio() + return max(token, seq) + + +def _enum_values(parameter: ToolParameter) -> set[str] | None: + values = parameter.schema_.get("enum") + if values is None or not isinstance(values, list): + return None + return {str(value) for value in values} + + +def _suggest_argument_renames( + old_tool: ToolContract, + new_tool: ToolContract, + *, + threshold: float = 0.5, +) -> tuple[dict[str, str], list[SuggestionNote]]: + old_only = {p.name: p for p in old_tool.parameters} + new_only = {p.name: p for p in new_tool.parameters} + shared = old_only.keys() & new_only.keys() + removed = {name: old_only[name] for name in old_only.keys() - shared} + added = {name: new_only[name] for name in new_only.keys() - shared} + pairs: list[tuple[float, str, str]] = [] + for old_name, old_param in removed.items(): + for new_name, new_param in added.items(): + score = _param_similarity(old_name, new_name) + # Slight boost when both are required or both optional. + if old_param.required == new_param.required: + score = min(1.0, score + 0.05) + if old_param.schema_.get("type") == new_param.schema_.get("type"): + score = min(1.0, score + 0.05) + if score >= threshold: + pairs.append((score, old_name, new_name)) + # Equal cardinality with no high-score pairs: greedy unique pairing. + if not pairs and len(removed) == len(added) and removed: + candidates: list[tuple[float, str, str]] = [] + for old_name in removed: + for new_name in added: + score = max(_param_similarity(old_name, new_name), 0.45) + candidates.append((score, old_name, new_name)) + candidates.sort(reverse=True) + used_old: set[str] = set() + used_new: set[str] = set() + for score, old_name, new_name in candidates: + if old_name in used_old or new_name in used_new: + continue + used_old.add(old_name) + used_new.add(new_name) + pairs.append((score, old_name, new_name)) + pairs.sort(reverse=True) + matched_old: set[str] = set() + matched_new: set[str] = set() + renames: dict[str, str] = {} + notes: list[SuggestionNote] = [] + for score, old_name, new_name in pairs: + if old_name in matched_old or new_name in matched_new: + continue + matched_old.add(old_name) + matched_new.add(new_name) + renames[old_name] = new_name + confidence = ( + SuggestionConfidence.HIGH + if score >= 0.75 + else SuggestionConfidence.MEDIUM + if score >= 0.55 + else SuggestionConfidence.LOW + ) + notes.append( + SuggestionNote( + subject=f"{old_tool.name}.{old_name}->{new_tool.name}.{new_name}", + confidence=confidence, + message=f"Parameter rename suggested (similarity={score:.2f}).", + ) + ) + # Sole leftover remove+add: low-confidence pairing. + leftover_old = sorted(removed.keys() - matched_old) + leftover_new = sorted(added.keys() - matched_new) + if len(leftover_old) == 1 and len(leftover_new) == 1: + old_name, new_name = leftover_old[0], leftover_new[0] + renames[old_name] = new_name + matched_old.add(old_name) + matched_new.add(new_name) + notes.append( + SuggestionNote( + subject=f"{old_tool.name}.{old_name}->{new_tool.name}.{new_name}", + confidence=SuggestionConfidence.LOW, + message="Parameter rename suggested as sole leftover remove+add pair.", + ) + ) + for old_name in sorted(removed.keys() - matched_old): + notes.append( + SuggestionNote( + subject=f"{old_tool.name}.{old_name}", + confidence=SuggestionConfidence.SKIPPED, + message="Removed parameter has no unambiguous rename target; skipped.", + ) + ) + return renames, notes + + +def _suggest_enum_map( + tool_name: str, + old_param: ToolParameter, + new_param: ToolParameter, +) -> tuple[dict[str, str] | None, SuggestionNote]: + old_vals = _enum_values(old_param) + new_vals = _enum_values(new_param) + if old_vals is None or new_vals is None: + return None, SuggestionNote( + subject=f"{tool_name}.{new_param.name}", + confidence=SuggestionConfidence.SKIPPED, + message="Enum remap skipped (one side is not an enum).", + ) + # Identity overlap only when names match exactly — unambiguous. + identity = old_vals & new_vals + if identity == old_vals == new_vals: + return None, SuggestionNote( + subject=f"{tool_name}.{new_param.name}", + confidence=SuggestionConfidence.SKIPPED, + message="Enum values unchanged; no remap needed.", + ) + # Unambiguous 1:1 only when equal cardinality and each old value has a + # unique best match above threshold with no ties. + if len(old_vals) != len(new_vals) or not old_vals: + return None, SuggestionNote( + subject=f"{tool_name}.{new_param.name}", + confidence=SuggestionConfidence.SKIPPED, + message=( + "Enum remap skipped (cardinality differs or empty); human authoring required." + ), + ) + mapping: dict[str, str] = {} + used_new: set[str] = set() + for old in sorted(old_vals): + ranked = sorted( + ( + (_jaccard(_token_set(old), _token_set(new)), new) + for new in new_vals + if new not in used_new + ), + reverse=True, + ) + if not ranked or ranked[0][0] < 0.5: + return None, SuggestionNote( + subject=f"{tool_name}.{new_param.name}", + confidence=SuggestionConfidence.SKIPPED, + message=f"No unambiguous enum target for '{old}'; skipped.", + ) + if len(ranked) > 1 and ranked[0][0] == ranked[1][0]: + return None, SuggestionNote( + subject=f"{tool_name}.{new_param.name}", + confidence=SuggestionConfidence.SKIPPED, + message=f"Tied enum targets for '{old}'; skipped.", + ) + mapping[old] = ranked[0][1] + used_new.add(ranked[0][1]) + return mapping, SuggestionNote( + subject=f"{tool_name}.{new_param.name}", + confidence=SuggestionConfidence.MEDIUM, + message="Enum value remap suggested from token similarity; review before use.", + ) + + +def suggest_adapter( + baseline: InterfaceSnapshot, + candidate: InterfaceSnapshot, + *, + report: CompatibilityReport | None = None, + rename_threshold: float = 0.55, + suggestion_rename_threshold: float = 0.4, +) -> AdapterSuggestion: + """Build a reviewable MigrationAdapter draft from baseline/candidate snapshots. + + Uses ``tool.renamed`` when present; may also propose aliases for remaining + remove+add pairs above ``suggestion_rename_threshold`` (marked low/medium + confidence). Parameter and enum maps are only emitted when unambiguous. + """ + report = report or compare_snapshots(baseline, candidate, detect_renames=True) + before = {tool.name: tool for tool in baseline.tools} + after = {tool.name: tool for tool in candidate.tools} + notes: list[SuggestionNote] = [] + aliases: list[ToolAlias] = [] + arguments: list[ArgumentMap] = [] + enums: list[EnumMap] = [] + + renamed_pairs: list[tuple[str, str]] = [] + for change in report.changes: + if change.code == "tool.renamed" and "->" in change.subject: + old_name, new_name = change.subject.split("->", 1) + renamed_pairs.append((old_name, new_name)) + aliases.append(ToolAlias(**{"from": old_name, "to": new_name})) + notes.append( + SuggestionNote( + subject=change.subject, + confidence=SuggestionConfidence.HIGH, + message="Alias from diff engine tool.renamed.", + ) + ) + + renamed_from = {old for old, _ in renamed_pairs} + renamed_to = {new for _, new in renamed_pairs} + remaining_removed = { + name: before[name] for name in before.keys() - after.keys() - renamed_from if name in before + } + remaining_added = { + name: after[name] for name in after.keys() - before.keys() - renamed_to if name in after + } + if remaining_removed and remaining_added: + extra = _detect_renames( + remaining_removed, + remaining_added, + threshold=suggestion_rename_threshold, + ) + # Sole remove+add of matching risk: low-confidence alias even when + # token similarity is weak (common intentional rename pattern). + if not extra and len(remaining_removed) == 1 and len(remaining_added) == 1: + old_name = next(iter(remaining_removed)) + new_name = next(iter(remaining_added)) + if remaining_removed[old_name].risk == remaining_added[new_name].risk: + extra = [(old_name, new_name)] + for old_name, new_name in extra: + score = _tool_similarity(before[old_name], after[new_name]) + if ( + score <= 0.0 + and (old_name, new_name) + in [(next(iter(remaining_removed)), next(iter(remaining_added)))] + and len(remaining_removed) == 1 + ): + confidence = SuggestionConfidence.LOW + detail = ( + "Alias suggested as sole remove+add pair with matching risk; " + "similarity was low — review carefully." + ) + else: + confidence = ( + SuggestionConfidence.HIGH + if score >= rename_threshold + else SuggestionConfidence.MEDIUM + if score >= suggestion_rename_threshold + else SuggestionConfidence.LOW + ) + detail = ( + f"Alias suggested from remove+add similarity ({score:.2f}); " + "not emitted as tool.renamed in the structural report." + ) + aliases.append(ToolAlias(**{"from": old_name, "to": new_name})) + renamed_pairs.append((old_name, new_name)) + notes.append( + SuggestionNote( + subject=f"{old_name}->{new_name}", + confidence=confidence, + message=detail, + ) + ) + + # Parameter / enum maps for renamed pairs and same-name tools. + pairs_to_inspect: list[tuple[str, ToolContract, ToolContract]] = [] + for old_name, new_name in renamed_pairs: + if old_name in before and new_name in after: + pairs_to_inspect.append((new_name, before[old_name], after[new_name])) + for name in sorted(before.keys() & after.keys()): + pairs_to_inspect.append((name, before[name], after[name])) + + for target_name, old_tool, new_tool in pairs_to_inspect: + renames, rename_notes = _suggest_argument_renames(old_tool, new_tool) + notes.extend(rename_notes) + if renames: + arguments.append(ArgumentMap(tool=target_name, rename=renames)) + # Enum maps on shared or renamed parameters. + old_params = {p.name: p for p in old_tool.parameters} + new_params = {p.name: p for p in new_tool.parameters} + for old_name, new_name in renames.items(): + if old_name in old_params and new_name in new_params: + mapping, note = _suggest_enum_map( + target_name, old_params[old_name], new_params[new_name] + ) + notes.append(note) + if mapping: + enums.append(EnumMap(tool=target_name, parameter=new_name, values=mapping)) + for shared in old_params.keys() & new_params.keys(): + mapping, note = _suggest_enum_map(target_name, old_params[shared], new_params[shared]) + notes.append(note) + if mapping: + enums.append(EnumMap(tool=target_name, parameter=shared, values=mapping)) + + # Document non-goals / skips for required-param adds without defaults. + for change in report.changes: + if change.code == "parameter.added_required": + notes.append( + SuggestionNote( + subject=change.subject, + confidence=SuggestionConfidence.SKIPPED, + message=( + "Required parameter added — no safe default; adapter suggestion skipped." + ), + ) + ) + + return AdapterSuggestion( + adapter=MigrationAdapter(aliases=aliases, arguments=arguments, enums=enums), + notes=notes, + baseline=report.baseline, + candidate=report.candidate, + auto_apply=False, + ) + + +def adapter_suggestion_to_json(suggestion: AdapterSuggestion) -> dict[str, Any]: + """Serialize suggestion for disk (adapter + notes; auto_apply always false).""" + return suggestion.model_dump(mode="json", by_alias=True) diff --git a/tests/test_suggest.py b/tests/test_suggest.py new file mode 100644 index 0000000..996e7ae --- /dev/null +++ b/tests/test_suggest.py @@ -0,0 +1,109 @@ +"""Tests for adapter suggestion drafts (#62).""" + +from __future__ import annotations + +import json +from pathlib import Path + +from typer.testing import CliRunner + +from tool_semantics.adapters import CompatibilityProxy +from tool_semantics.cli import app +from tool_semantics.diff import compare_snapshots +from tool_semantics.models import InterfaceSnapshot, ToolContract, ToolParameter +from tool_semantics.scanner import capture_manifest, write_snapshot +from tool_semantics.suggest import SuggestionConfidence, suggest_adapter + + +def test_suggest_adapter_github_v1_v2() -> None: + baseline = capture_manifest(Path("examples/github_server_v1.json")) + candidate = capture_manifest(Path("examples/github_server_v2.json")) + suggestion = suggest_adapter(baseline, candidate) + assert suggestion.auto_apply is False + # search_issues → find_work_items via suggestion-threshold rename + assert any(alias.from_name == "search_issues" for alias in suggestion.adapter.aliases) + assert any(alias.to_name == "find_work_items" for alias in suggestion.adapter.aliases) + # query → search_expression and/or state → status when similarity allows + rename_maps = [item.rename for item in suggestion.adapter.arguments] + flat = {k: v for mapping in rename_maps for k, v in mapping.items()} + assert "query" in flat or "state" in flat + # Required repository add is skipped (no safe default) + assert any( + note.confidence == SuggestionConfidence.SKIPPED and "repository" in note.subject + for note in suggestion.notes + ) + + +def test_suggest_adapter_from_tool_renamed_and_enum() -> None: + baseline = InterfaceSnapshot( + server_name="demo", + server_version="1", + tools=[ + ToolContract( + name="list_items", + description="List work items by status filter", + parameters=[ + ToolParameter( + name="status", + schema={"type": "string", "enum": ["open", "closed"]}, + required=True, + ) + ], + ) + ], + ) + candidate = InterfaceSnapshot( + server_name="demo", + server_version="2", + tools=[ + ToolContract( + name="list_work_items", + description="List work items by status filter", + parameters=[ + ToolParameter( + name="status", + schema={"type": "string", "enum": ["open", "closed", "archived"]}, + required=True, + ) + ], + ) + ], + ) + report = compare_snapshots(baseline, candidate) + suggestion = suggest_adapter(baseline, candidate, report=report) + assert suggestion.adapter.aliases + # Cardinality differs → enum remap skipped + assert any( + note.confidence == SuggestionConfidence.SKIPPED and "enum" in note.message.lower() + for note in suggestion.notes + ) + + +def test_suggested_adapter_is_usable_via_proxy() -> None: + baseline = capture_manifest(Path("examples/github_server_v1.json")) + candidate = capture_manifest(Path("examples/github_server_v2.json")) + suggestion = suggest_adapter(baseline, candidate) + proxy = CompatibilityProxy(suggestion.adapter) + tool, args = proxy.route_call("search_issues", {"query": "bug", "state": "open"}) + assert tool == "find_work_items" + # At least one renamed arg should land under the modern name when mapped + assert "query" not in args or "search_expression" in args or args.get("query") == "bug" + + +def test_suggest_adapter_cli(tmp_path: Path) -> None: + baseline = capture_manifest(Path("examples/github_server_v1.json")) + candidate = capture_manifest(Path("examples/github_server_v2.json")) + base_path = tmp_path / "v1.json" + cand_path = tmp_path / "v2.json" + write_snapshot(baseline, base_path) + write_snapshot(candidate, cand_path) + out = tmp_path / "adapter.json" + result = CliRunner().invoke( + app, + ["suggest-adapter", str(base_path), str(cand_path), "-o", str(out)], + ) + assert result.exit_code == 0, result.output + payload = json.loads(out.read_text(encoding="utf-8")) + assert payload["auto_apply"] is False + assert "adapter" in payload + assert "notes" in payload