feat(mcp): surface alias keys a documented data-shape omits (#780)

The documented-shape path answers "what fields does <blob> contain" at high
confidence and tells the caller no verification Read is needed. But a docstring
lists the declared shape, which can be a subset of what the records actually
carry: consumers often read a legacy alias the doc never mentions. For
co_change_partners_json, four consumers defensively read
`partner.get("co_change_count") or partner.get("count")` and
`... or partner.get("path")`, so `count` and `path` are real keys the documented
shape omits. Answering high-confidence "cite it, no Read needed" while hiding
those is a confidently-incomplete answer.

Mine them and surface them. After the documented shape is chosen, scan the
identifier's consumer files for the alias idiom `<recv>.get("<A>") or
<recv>.get("<B>")` where one key is a documented field; the other is an alias for
it. Requiring the `or` fallback on a shared receiver keeps this tight: an
assignment that merely co-mentions a documented key on an unrelated record, or a
test assertion, does not match. The extras ride back in a new
`data_shape.also_accessed` block (field + file:line), and the answer text names
them so the agent handles them instead of trusting a subset. The documented
fields stay authoritative; confidence stays high on the declared core.

Adds three tests: an omitted `or`-fallback alias is surfaced, a clean shape with
no divergence carries no also_accessed, and a cross-record co-mention (an
assignment, not a fallback) is correctly not reported as an alias.

Author: RaghavChamadiya

packages/server/src/repowise/server/mcp_server/tool_answer/answer.py

@@ -303,21 +303,43 @@ def _build_data_shape_payload(grounded: dict, t0: float, repository) -> dict:
     ident = grounded["identifier"]
     fields = grounded["fields"]
     sources = grounded["sources"]
+    also_accessed = grounded.get("also_accessed") or []
     citations = sorted({s["file"] for s in sources})
     field_list = ", ".join(f"`{f}`" for f in fields)
     doc_src = next((s for s in sources if s["kind"] == "docstring"), None)
     if grounded["grounding"] == "docstring":
         where = f"{doc_src['file']}:{doc_src['line']}" if doc_src else citations[0]
-        answer = (
-            f"Each entry in `{ident}` has {len(fields)} field(s): {field_list}. "
-            f"This is the documented shape at {where}; cite it directly, no "
-            "verification Read needed."
-        )
-        note = (
-            "Grounded in the documented field shape mined from source (the "
-            "quoted keys in the docstring/comment at the cited line). "
-            "data_shape.sources lists every field's origin line."
-        )
+        if also_accessed:
+            # The doc lists the declared shape, but consumers read alias key(s)
+            # it omits (a legacy fallback). Surface them: telling the agent "no
+            # Read needed" while hiding a key it must handle would be a
+            # confidently-incomplete answer.
+            alias_list = ", ".join(f"`{a['field']}`" for a in also_accessed)
+            first_alias = also_accessed[0]
+            answer = (
+                f"Each entry in `{ident}` has {len(fields)} documented field(s): "
+                f"{field_list} (documented shape at {where}). Consumers also read "
+                f"{alias_list} as a fallback (e.g. {first_alias['file']}:"
+                f"{first_alias['line']}) - an alias the docstring omits, so handle "
+                f"{alias_list} too if you touch this."
+            )
+            note = (
+                "Grounded in the documented field shape, plus alias key(s) "
+                "consumers read beside a documented field that the docstring "
+                "omits (see data_shape.also_accessed). The documented fields are "
+                "authoritative; the aliases are real keys the code defends against."
+            )
+        else:
+            answer = (
+                f"Each entry in `{ident}` has {len(fields)} field(s): {field_list}. "
+                f"This is the documented shape at {where}; cite it directly, no "
+                "verification Read needed."
+            )
+            note = (
+                "Grounded in the documented field shape mined from source (the "
+                "quoted keys in the docstring/comment at the cited line). "
+                "data_shape.sources lists every field's origin line."
+            )
     else:
         first = sources[0]
         answer = (
@@ -340,6 +362,7 @@ def _build_data_shape_payload(grounded: dict, t0: float, repository) -> dict:
             "identifier": ident,
             "fields": fields,
             "sources": sources,
+            **({"also_accessed": also_accessed} if also_accessed else {}),
         },
         "fallback_targets": citations,
         "retrieval": [],

packages/server/src/repowise/server/mcp_server/tool_answer/data_shape.py

@@ -142,14 +142,29 @@ def _order_candidates(files: list[str]) -> list[str]:
     )
 
 
-def _run_grep(repo_root: Path, args: list[str], identifier: str) -> subprocess.CompletedProcess | None:
+def _run_grep(
+    repo_root: Path, args: list[str], identifier: str
+) -> subprocess.CompletedProcess | None:
     """Run a bounded, transport-safe ``git grep`` variant; ``None`` on failure."""
     try:
         return subprocess.run(
             # --no-pager + stdin=DEVNULL are load-bearing: this can run inside a
             # stdio MCP server whose stdin IS the JSON-RPC pipe, so a pager that
             # reads stdin would deadlock the transport.
-            ["git", "--no-pager", "grep", *args, "-l", "-I", "-F", "-w", "-e", identifier, "--", *_GREP_PATHSPECS],
+            [
+                "git",
+                "--no-pager",
+                "grep",
+                *args,
+                "-l",
+                "-I",
+                "-F",
+                "-w",
+                "-e",
+                identifier,
+                "--",
+                *_GREP_PATHSPECS,
+            ],
             cwd=str(repo_root),
             capture_output=True,
             text=True,
@@ -259,6 +274,40 @@ def _doc_shape_in_file(
     return None
 
 
+# A ``<receiver>.get("<key>")`` access - group(1) receiver, group(2) key.
+_GET_ACCESS = re.compile(r"([A-Za-z_]\w*)\s*\.\s*get\(\s*['\"]([A-Za-z_]\w*)['\"]")
+
+
+def _alias_keys_on_documented_lines(
+    lines: list[str], doc_fields: set[str]
+) -> list[tuple[str, int]]:
+    """Alias keys a documented field is read as a fallback for.
+
+    Targets one idiom precisely: ``<recv>.get("<A>") or <recv>.get("<B>")`` - the
+    same receiver reads two keys joined by ``or``, so when one is a documented
+    field the other is an alias for it (``partner.get("co_change_count") or
+    partner.get("count")`` -> ``count``; ``... or partner.get("path")`` ->
+    ``path``). Requiring the ``or`` fallback and a shared receiver keeps this tight:
+    an assignment that merely co-mentions a documented key on a different record
+    (``meta["prior_defect_count"] = ...meta["file_path"]``) or a test assertion
+    does not match. Returns ``(alias, line)`` for keys not in ``doc_fields``.
+    """
+    out: list[tuple[str, int]] = []
+    for idx, line in enumerate(lines, 1):
+        if " or " not in line:
+            continue
+        by_recv: dict[str, list[str]] = {}
+        for m in _GET_ACCESS.finditer(line):
+            by_recv.setdefault(m.group(1), []).append(m.group(2))
+        for keys in by_recv.values():
+            if len(keys) < 2 or not any(k in doc_fields for k in keys):
+                continue
+            for k in keys:
+                if k not in doc_fields:
+                    out.append((k, idx))
+    return out
+
+
 def _accessed_fields_in_file(
     lines: list[str], identifier: str, mentions: list[int]
 ) -> list[tuple[str, int]]:
@@ -323,6 +372,9 @@ def mine_data_shape(repo_root: Path | None, question_ids: set[str]) -> dict | No
           "grounding": "docstring" | "access",
           "confidence": "high" | "medium",
           "sources": [{"file", "line", "kind"}],   # kind in {docstring, access}
+          # docstring grounding only, and only when present: keys consumers read
+          # beside a documented field that the doc omits (aliases / optional keys)
+          "also_accessed": [{"field", "file", "line"}],
         }
     """
     if repo_root is None:
@@ -344,6 +396,7 @@ def mine_data_shape(repo_root: Path | None, question_ids: set[str]) -> dict | No
         access_fields: list[str] = []
         access_sources: list[dict] = []
         access_seen: set[str] = set()
+        file_lines: dict[str, list[str]] = {}  # cached for the divergence pass
 
         for rel in files:
             lines = _read_lines(root, rel)
@@ -352,6 +405,7 @@ def mine_data_shape(repo_root: Path | None, question_ids: set[str]) -> dict | No
             mentions = _mention_lines(lines, identifier)
             if not mentions:
                 continue
+            file_lines[rel] = lines
             doc = _doc_shape_in_file(lines, identifier, mentions)
             if doc is not None:
                 fields, line = doc
@@ -387,13 +441,30 @@ def _rank(item: tuple[frozenset, list[tuple[list[str], str, int]]]):
                 if any(src["file"] == s["file"] for s in sources):
                     continue
                 sources.append(src)
-            return {
+            # Divergence: keys consumers read right beside a documented field but
+            # the doc never lists (a legacy alias like ``count`` for
+            # ``co_change_count``, an optional key). The documented shape is
+            # authoritative for what it declares, but if we said "cite it, no Read
+            # needed" while hiding a key four consumers defensively handle, an
+            # agent could ship a change that ignores it. Surface it instead.
+            doc_field_set = set(fields)
+            also_accessed: list[dict] = []
+            also_seen: set[str] = set()
+            for rel, lines in file_lines.items():
+                for key, line in _alias_keys_on_documented_lines(lines, doc_field_set):
+                    if key not in also_seen:
+                        also_seen.add(key)
+                        also_accessed.append({"field": key, "file": rel, "line": line})
+            result: dict = {
                 "identifier": identifier,
                 "fields": fields,
                 "grounding": "docstring",
                 "confidence": "high",
                 "sources": sources,
             }
+            if also_accessed:
+                result["also_accessed"] = also_accessed
+            return result
 
         # No documented shape - fall back to consistent key accesses.
         if len(access_fields) >= _DATA_SHAPE_MIN_FIELDS:

tests/unit/server/mcp/test_answer_data_shape.py

@@ -237,5 +237,103 @@ def consume(event_payload_json):
     assert out["fields"] == ["kind", "ts", "actor"]
 
 
+# --- Alias divergence (doc omits a key consumers read as a fallback) -------
+
+
+def test_documented_shape_surfaces_alias_fallback(tmp_path):
+    """A doc shape that omits an ``or``-fallback alias must surface the alias.
+
+    The docstring is authoritative for what it declares, but if a consumer reads
+    ``x.get("weight_json") or x.get("weight")`` the ``weight`` alias is a real key
+    the tool must not hide behind "no verification needed".
+    """
+    _write(
+        tmp_path,
+        "pkg/models.py",
+        '''\
+class M:
+    """``metric_rows_json`` is a JSON list of
+    ``{"file_path", "weight_json", "ts"}`` metric records."""
+
+    metric_rows_json: str
+''',
+    )
+    _write(
+        tmp_path,
+        "pkg/consume.py",
+        """\
+def use(metric_rows_json):
+    for row in parse(metric_rows_json):
+        w = row.get("weight_json") or row.get("weight") or 0
+    return w
+""",
+    )
+    out = mine_data_shape(tmp_path, {"metric_rows_json"})
+    assert out is not None
+    assert out["grounding"] == "docstring"
+    assert out["fields"] == ["file_path", "weight_json", "ts"]
+    aliases = {a["field"] for a in out.get("also_accessed", [])}
+    assert aliases == {"weight"}
+    assert out["also_accessed"][0]["file"] == "pkg/consume.py"
+
+
+def test_no_alias_key_when_none_diverges(tmp_path):
+    """A clean documented shape with no fallback alias carries no also_accessed."""
+    _write(
+        tmp_path,
+        "pkg/clean.py",
+        '''\
+class M:
+    """``clean_rows_json`` is a JSON list of
+    ``{"file_path", "score", "ts"}`` records."""
+
+    clean_rows_json: str
+
+
+def use(clean_rows_json):
+    for r in parse(clean_rows_json):
+        s = r.get("score")
+    return s
+''',
+    )
+    out = mine_data_shape(tmp_path, {"clean_rows_json"})
+    assert out is not None
+    assert "also_accessed" not in out
+
+
+def test_alias_precision_ignores_cross_record_comention(tmp_path):
+    """A documented key co-mentioned on a different record is NOT an alias.
+
+    ``meta["other_count"] = src[meta["file_path"]]`` reads a documented field
+    (``file_path``) and another key on the same line, but it's an assignment on an
+    unrelated record, not an ``or`` fallback - it must not be reported as an alias.
+    """
+    _write(
+        tmp_path,
+        "pkg/models.py",
+        '''\
+class M:
+    """``rows_json`` is a JSON list of
+    ``{"file_path", "amount", "ts"}`` records."""
+
+    rows_json: str
+''',
+    )
+    _write(
+        tmp_path,
+        "pkg/enrich.py",
+        """\
+def enrich(rows_json, meta, src):
+    for row in parse(rows_json):
+        amount = row.get("amount")
+    meta["other_count"] = src[meta["file_path"]]
+    return amount, meta
+""",
+    )
+    out = mine_data_shape(tmp_path, {"rows_json"})
+    assert out is not None
+    assert "also_accessed" not in out
+
+
 def test_none_repo_root_is_safe():
     assert mine_data_shape(None, {"anything_json"}) is None