add trace hints and skill integration (PR-TRACE-3) (#248)

* add trace hints and skill integration (PR-TRACE-3)

- Extend generate_hints with output_kind="trace" and four hint templates
  (budget hit, pruned drilldown, cross-service boundary, high fanout trace)
- Add trace recommendation to neighbors/describe high-fanout hints
- Update SKILL.md preamble, decision tree, and tool reference with trace
- Add "trace" to StructuredHint.tool Literal in mcp_v2.py
- Add 5 new tests (4 hint unit tests + 1 integration test)

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* address PR-TRACE-3 review feedback

- Extract _high_fanout_trace_hint() helper (review point 1)
- Use user's prune_roles in budget-hit hint if supplied (point 2)
- Include from_id in cross-service hint reason (point 3)
- Split stats guard so cross-service hints work without stats (point 4)
- Add clarifying comment for fan_out_cap=0 in integration test (point 5)

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>

Author: HumanBean17

mcp_hints.py

@@ -133,6 +133,10 @@ def _extract_other_ids(results: list[dict[str, Any]]) -> list[str]:
 LABEL_WRONG_SUBJECT_KIND = "wrong subject kind"
 LABEL_WRONG_DIRECTION = "wrong direction"
 LABEL_TYPE_LEVEL_REQUERY = "type-level requery"
+LABEL_BUDGET_HIT = "budget hit"
+LABEL_PRUNED_DRILLDOWN = "pruned drilldown"
+LABEL_CROSS_SERVICE_BOUNDARY = "cross-service boundary"
+LABEL_HIGH_FANOUT_TRACE = "high fanout trace"
 
 # §7.12 priority: DECLARES.* type rollups > OVERRIDDEN_BY.* > leaf follow-ups > meta.
 PRIORITY_DECLARES_TYPE_ROLLUP = 4
@@ -501,8 +505,95 @@ def _neighbors_success_structured_hints(payload: dict[str, Any]) -> list[_Struct
     return out
 
 
+def _high_fanout_trace_hint(origin_id: str, calls_n: int) -> _StructuredHint:
+    """Shared trace recommendation for high CALLS fan-out (used by neighbors and describe paths)."""
+    return _StructuredHint(
+        "trace",
+        {"ids": [origin_id], "direction": "out", "edge_types": ["CALLS"],
+         "prune_roles": ["DTO", "EXCEPTION", "UTILITY"], "fan_out_cap": 5},
+        True, PRIORITY_LEAF_FOLLOWUP,
+        LABEL_HIGH_FANOUT_TRACE,
+        f"{calls_n} CALLS — consider trace for a pruned multi-hop view",
+    )
+
+
+def _trace_structured_hints(payload: dict[str, Any]) -> tuple[list[_StructuredHint], list[tuple[int, str]]]:
+    """Structured hints and advisories for trace output."""
+    struct_pairs: list[_StructuredHint] = []
+    advisories: list[tuple[int, str]] = []
+
+    stats = payload.get("stats")
+    edges = list(payload.get("edges") or [])
+
+    # Cross-service boundary hints don't require stats — guard stats-dependant hints separately.
+    if isinstance(stats, dict):
+        # (a) Budget hit hint.
+        if stats.get("budget_hit"):
+            n = int(stats.get("total_nodes_discovered") or 0)
+            advisories.append((
+                PRIORITY_META,
+                f"trace hit the node discovery budget ({n} nodes). "
+                "Results are partial. Increase max_depth or add prune_roles and re-run.",
+            ))
+            # Preserve user's prune_roles if they supplied custom ones.
+            prune_roles = list(payload.get("prune_roles") or ["DTO", "EXCEPTION", "UTILITY"])
+            struct_pairs.append(_StructuredHint(
+                "trace",
+                {"ids": list(payload.get("seed_ids") or []),
+                 "direction": str(payload.get("direction") or "out"),
+                 "edge_types": list(payload.get("edge_types") or []),
+                 "prune_roles": prune_roles},
+                False, PRIORITY_META,
+                LABEL_BUDGET_HIT,
+                f"budget hit at {n} nodes — results are incomplete",
+            ))
+
+        # (b) Pruned / collapsed edges drill-down hint.
+        pruned_count = (
+            int(stats.get("nodes_pruned_role") or 0)
+            + int(stats.get("nodes_pruned_fan_out") or 0)
+            + int(stats.get("edges_collapsed_trivial") or 0)
+        )
+        if pruned_count > 0 or any(e.get("collapsed") for e in edges):
+            advisories.append((
+                PRIORITY_META,
+                f"trace pruned {pruned_count} edges. Use neighbors(id, direction, edge_types) on specific nodes for full detail.",
+            ))
+            struct_pairs.append(_StructuredHint(
+                "neighbors",
+                {"ids": [], "direction": str(payload.get("direction") or "out"), "edge_types": list(payload.get("edge_types") or [])},
+                False, PRIORITY_META,
+                LABEL_PRUNED_DRILLDOWN,
+                f"trace pruned {pruned_count} edges — use neighbors on specific nodes for detail",
+            ))
+
+    # (c) Cross-service boundary hint (no stats dependency).
+    xs_edges = [e for e in edges if e.get("cross_service_boundary")]
+    if xs_edges:
+        for xe in xs_edges[:3]:
+            to_id = str(xe.get("to_id") or "")
+            from_id = str(xe.get("from_id") or "")
+            confidence = xe.get("attrs", {}).get("confidence") if isinstance(xe.get("attrs"), dict) else None
+            conf_str = f"confidence={confidence}" if confidence is not None else "low confidence"
+            advisories.append((
+                PRIORITY_META,
+                f"Cross-service boundary: {from_id} -> {to_id} ({conf_str}). "
+                f"Use trace('{to_id}', 'out', ['EXPOSES','CALLS'], max_depth=4) to continue in the downstream service, "
+                f"or describe('{to_id}') for route details.",
+            ))
+            struct_pairs.append(_StructuredHint(
+                "trace",
+                {"ids": [to_id], "direction": "out", "edge_types": ["EXPOSES", "CALLS"], "max_depth": 4},
+                True, PRIORITY_LEAF_FOLLOWUP,
+                LABEL_CROSS_SERVICE_BOUNDARY,
+                f"cross-service boundary: {from_id} -> {to_id}",
+            ))
+
+    return (finalize_structured_hints(struct_pairs), finalize_advisories(advisories))
+
+
 def generate_hints(
-    output_kind: Literal["search", "find", "describe", "neighbors", "resolve"],
+    output_kind: Literal["search", "find", "describe", "neighbors", "resolve", "trace"],
     payload: dict[str, Any],
 ) -> tuple[list[_StructuredHint], list[str]]:
     """Return structured hints and advisories for a success-only MCP v2 payload dict.
@@ -685,6 +776,8 @@ def generate_hints(
                             LABEL_HIGH_FANOUT,
                             f"{calls_n} CALLS — noisy axes are callee_declaring_role and per-call-site multiplicity",
                         ))
+                        # Trace recommendation for high fan-out neighbors results.
+                        struct_meta.append(_high_fanout_trace_hint(origin_id, calls_n))
             # Unresolved sites advisory
             unresolved = int(payload.get("unresolved_count") or 0)
             if unresolved > 0:
@@ -925,8 +1018,16 @@ def generate_hints(
                     LABEL_HIGH_FANOUT,
                     f"method has {_out_count(edge_summary, 'CALLS')} CALLS — consider filtering",
                 ))
+                struct_pairs.append(_high_fanout_trace_hint(node_id, _out_count(edge_summary, "CALLS")))
                 # Advisory for many CALLS
                 advisories.append((PRIORITY_LEAF_FOLLOWUP, "many CALLS — consider filtering by target microservice"))
             return (finalize_structured_hints(struct_pairs), finalize_advisories(advisories))
 
         return ([], [])
+
+    if output_kind == "trace":
+        if not payload.get("success"):
+            return ([], [])
+        return _trace_structured_hints(payload)
+
+    return ([], [])

mcp_v2.py

@@ -178,7 +178,7 @@ def _role_axes_mutually_exclusive(self) -> EdgeFilter:
 
 class StructuredHint(BaseModel):
     label: str = ""
-    tool: Literal["search", "find", "describe", "neighbors", "resolve"]
+    tool: Literal["search", "find", "describe", "neighbors", "trace", "resolve"]
     args: dict[str, Any]
     actionable: bool = True
     reason: str = ""

skills/explore-codebase/SKILL.md

@@ -1,6 +1,6 @@
 ---
 name: explore-codebase
-description: Complete operating manual for the java-codebase-rag MCP tools (search, find, describe, neighbors, resolve). Use this skill whenever you need to explore a Java codebase — locate symbols, trace call chains, find routes, walk cross-service boundaries, or answer any "where is X", "who calls Y", "what does Z depend on" question. Self-contained: includes edge taxonomy, NodeFilter reference, decision tree, argument shapes, recovery playbook, and navigation patterns. No external files needed.
+description: Complete operating manual for the java-codebase-rag MCP tools (search, find, describe, neighbors, trace, resolve). Use this skill whenever you need to explore a Java codebase — locate symbols, trace call chains, find routes, walk cross-service boundaries, or answer any "where is X", "who calls Y", "what does Z depend on" question. Self-contained: includes edge taxonomy, NodeFilter reference, decision tree, argument shapes, recovery playbook, and navigation patterns. No external files needed.
 ---
 
 # /explore-codebase — Codebase navigation via the java-codebase-rag MCP
@@ -11,7 +11,7 @@ Any time you need to understand structure in an indexed Java codebase: locating
 
 ## Tools
 
-`search`, `find`, `describe`, `neighbors`, `resolve`.
+`search`, `find`, `describe`, `neighbors`, `trace`, `resolve`.
 
 ## Node kinds
 
@@ -47,15 +47,15 @@ Trust the indexed brownfield row over a framework-only reading of the source.
 
 1. **Locate** — `resolve` for identifier-shaped strings; `search` for natural language or code fragments; `find` for structured `NodeFilter` discovery.
 2. **Inspect** — `describe(id)` for the full record and `edge_summary` (per-label `in`/`out` counts).
-3. **Walk** — `neighbors` in a loop with explicit **`direction`** and **`edge_types`**. Multi-hop traces are **your** reasoning, not a separate tool.
+3. **Walk** — `neighbors` for single-hop adjacency; `trace` for multi-hop BFS with pruning (fan-out control, role-based pruning, cross-service boundaries).
 
 ## Forced reasoning preamble (every tool call)
 
 Before each MCP call, output one short line:
 
 ```
-Q-class: <semantic | structured | inspect | walk>
-Pick: <search|find|describe|neighbors|resolve>  Why: <≤8 words>
+Q-class: <semantic | structured | inspect | walk | trace>
+Pick: <search|find|describe|neighbors|trace|resolve>  Why: <≤8 words>
 ```
 
 Then use real JSON shapes (see below). If the call fails or returns nothing useful, use the **Recovery playbook** — do not thrash.
@@ -189,6 +189,10 @@ Prefer **`resolve` -> `describe(id=...)`** over **`describe(fqn=...)`** when an
 | Who implements interface T? | type symbol id | `neighbors(ids, "in", ["IMPLEMENTS"])` |
 | Where is T injected | type symbol id | `neighbors(ids, "in", ["INJECTS"])` |
 | Impact / "what breaks if I change X"? | no magic tool | loop `neighbors` `in` with `CALLS`, `INJECTS`, ... until bounded |
+| "What happens when route R is called?" | `find(kind="route")` then `trace(route_id, "out", ["EXPOSES","CALLS"], max_depth=4)` | `describe` on key nodes |
+| "Impact of changing method M" | `resolve` / `find` then `trace(id, "in", ["CALLS","OVERRIDES"], max_depth=3)` | `describe` on callers |
+| "Trace from X to database" | `trace(id, "out", ["CALLS"], max_depth=4, prune_roles=["DTO","EXCEPTION"])` | `neighbors` for pruned detail |
+| "What calls this across services?" | `trace(id, "out", ["CALLS","HTTP_CALLS","ASYNC_CALLS"], max_depth=5)` | `trace` on downstream route_id if needed |
 
 **Rules of thumb:**
 
@@ -229,6 +233,19 @@ Returns **edges** with `attrs` (`confidence`, `strategy`, `match`, ... on cross-
 
 **`CALLS` edges:** source-ordered (`call_site_line`, `call_site_byte`). `attrs.resolved=false` on `CALLS` rows means known-receiver-external (JDK/Spring) callees. **`include_unresolved=True`** (CALLS + `direction=out` only) interleaves unresolved sites with resolved `CALLS`. **`dedup_calls=True`** collapses identical `(origin, callee)` `CALLS` to one row with `call_site_lines`. Optional **`edge_filter`** projects before pagination: `min_confidence`; `include_strategies` / `exclude_strategies` (mutually exclusive); `callee_declaring_role`, `callee_declaring_roles`, `exclude_callee_declaring_roles` (`["OTHER"]` also drops known-external rows).
 
+### `trace`
+
+Multi-hop BFS with pruning. Args: `ids` (string or list), **`direction`**, **`edge_types`** (stored labels only — no composed dot-keys), `max_depth` (default 3, clamped 1–5), `max_paths` (default 20), `max_nodes_discovered` (default 500, clamped 100–2000), optional `filter` (NodeFilter), optional `edge_filter` (CALLS only), optional `prune_roles` (soft gate — edges recorded, frontier stops), `fan_out_cap` (default 5, scaffolding edges exempt), `collapse_trivial` (default true), `include_unresolved` (default false).
+
+Returns `TraceOutput`: `success`, `seed_ids`, `direction`, `edge_types`, `actual_depth`, `nodes` (dict of id→NodeRef), `edges` (list of `TraceEdge`), `paths` (list of `TracePath`), `stats` (`TraceStats`), `advisories`.
+
+**`TraceEdge`**: `from_id`, `to_id`, `edge_type`, `hop` (BFS depth), `parent_edge_id` (nullable), `collapsed` (bool), `collapsed_intermediates` (list of node ids), `cross_service_boundary` (bool), `attrs`.
+
+**When to use `trace` vs `neighbors`:**
+- Use `neighbors` for single-hop adjacency where you want the full unfiltered result.
+- Use `trace` for multi-hop path questions (3+ hops), impact analysis, cross-service boundary discovery, or when `neighbors` returns high fan-out (>8 CALLS edges).
+- After `trace`, use `neighbors` or `describe` on specific nodes for detail the trace pruned or collapsed.
+
 ## Ontology glossary
 
 **Roles (`filter.role` / `exclude_roles`):** `CONTROLLER`, `SERVICE`, `REPOSITORY`, `COMPONENT`, `CONFIG`, `ENTITY`, `CLIENT`, `MAPPER`, `DTO`, `OTHER`.

tests/test_mcp_hints.py

@@ -1307,3 +1307,97 @@ def test_advisories_char_cap() -> None:
     _, advisories = generate_hints("neighbors", payload)
     for a in advisories:
         assert len(a) <= 200, f"advisory too long: {a!r}"
+
+
+# ---------------------------------------------------------------------------
+# PR-TRACE-3 tests: trace hints + neighbors high-fanout trace mention
+# ---------------------------------------------------------------------------
+
+
+def test_hint_trace_budget_hit() -> None:
+    """generate_hints('trace', {stats.budget_hit=True}) returns budget hit advisory."""
+    struct, advisories = generate_hints("trace", {
+        "success": True,
+        "stats": {
+            "budget_hit": True,
+            "total_nodes_discovered": 500,
+            "nodes_after_pruning": 120,
+        },
+        "edges": [],
+        "nodes": {},
+        "seed_ids": ["sym:com.example.Svc#run()"],
+        "direction": "out",
+        "edge_types": ["CALLS"],
+    })
+    assert any("budget" in a.lower() for a in advisories), f"no budget advisory in {advisories}"
+    budget_hints = [h for h in struct if h.label == "budget hit"]
+    assert budget_hints, f"no budget hit structured hint in {[h.label for h in struct]}"
+
+
+def test_hint_trace_pruned_edges() -> None:
+    """generate_hints('trace', {stats with pruning}) returns drill-down hint."""
+    struct, advisories = generate_hints("trace", {
+        "success": True,
+        "stats": {
+            "budget_hit": False,
+            "nodes_pruned_role": 5,
+            "nodes_pruned_fan_out": 3,
+            "edges_collapsed_trivial": 2,
+            "total_nodes_discovered": 100,
+        },
+        "edges": [],
+        "nodes": {},
+        "seed_ids": ["sym:a"],
+        "direction": "out",
+        "edge_types": ["CALLS"],
+    })
+    assert any("pruned" in a.lower() for a in advisories), f"no pruned advisory in {advisories}"
+    drilldown = [h for h in struct if h.label == "pruned drilldown"]
+    assert drilldown, f"no pruned drilldown hint in {[h.label for h in struct]}"
+
+
+def test_hint_trace_cross_service_boundary() -> None:
+    """generate_hints('trace', {edges with cross_service_boundary=True}) returns cross-service hint."""
+    struct, advisories = generate_hints("trace", {
+        "success": True,
+        "stats": {
+            "budget_hit": False,
+            "total_nodes_discovered": 10,
+            "nodes_pruned_role": 0,
+            "nodes_pruned_fan_out": 0,
+            "edges_collapsed_trivial": 0,
+        },
+        "edges": [
+            {
+                "from_id": "client:svc:PaymentClient",
+                "to_id": "route:payment-service:/api/payments:POST",
+                "cross_service_boundary": True,
+                "attrs": {"confidence": 0.85, "strategy": "URI_PATH_MATCH"},
+            },
+        ],
+        "nodes": {
+            "route:payment-service:/api/payments:POST": {"id": "route:payment-service:/api/payments:POST", "kind": "route"},
+        },
+        "seed_ids": ["sym:a"],
+        "direction": "out",
+        "edge_types": ["CALLS", "HTTP_CALLS"],
+    })
+    assert any("cross-service" in a.lower() for a in advisories), f"no cross-service advisory in {advisories}"
+    xs_hints = [h for h in struct if h.label == "cross-service boundary"]
+    assert xs_hints, f"no cross-service boundary hint in {[h.label for h in struct]}"
+    assert xs_hints[0].tool == "trace"
+    assert xs_hints[0].actionable is True
+
+
+def test_hint_neighbors_high_fanout_mentions_trace() -> None:
+    """generate_hints('neighbors', {many CALLS}) includes a trace recommendation."""
+    payload = _neighbors_hint_payload(
+        [_success_edge(_symbol_other(f"sym:callee{i}"), edge_type="CALLS") for i in range(12)],
+        requested_edge_types=["CALLS"],
+    )
+    payload["calls_row_count"] = 12
+    struct, _ = generate_hints("neighbors", payload)
+    trace_hints = [h for h in struct if h.tool == "trace" and h.label == "high fanout trace"]
+    assert trace_hints, f"no trace fanout hint in {[(h.tool, h.label) for h in struct]}"
+    assert trace_hints[0].actionable is True
+    assert "prune_roles" in trace_hints[0].args

tests/test_mcp_trace.py

@@ -948,3 +948,55 @@ async def test_trace_tool_description_mentions_six_tools(mcp_server) -> None:
     assert "describe" in instructions
     assert "neighbors" in instructions
     assert "resolve" in instructions
+
+
+# ---------------------------------------------------------------------------
+# PR-TRACE-3 tests: cross-service integration + hint verification
+# ---------------------------------------------------------------------------
+
+
+def test_trace_bank_chat_cross_service_http_flow(kuzu_graph: KuzuGraph) -> None:
+    """Integration: trace from a bank-chat method through HTTP_CALLS; verify cross-service boundary + hints."""
+    # Find a method that declares a client (has cross-service path).
+    seed_id = _find_method_with_declares_client(kuzu_graph)
+    if seed_id is None:
+        pytest.skip("No method with DECLARES_CLIENT in fixture")
+
+    out = trace_v2(
+        ids=seed_id,
+        direction="out",
+        edge_types=["CALLS", "HTTP_CALLS"],
+        max_depth=4,
+        fan_out_cap=0,  # No cap — include all edges for cross-service verification.
+        graph=kuzu_graph,
+    )
+    assert out.success is True
+
+    # Verify cross-service boundary edges exist.
+    xs_edges = [e for e in out.edges if e.cross_service_boundary]
+    if xs_edges:
+        # Cross-service boundary edges should stop at the route.
+        for xe in xs_edges:
+            assert xe.edge_type in ("HTTP_CALLS", "ASYNC_CALLS")
+            assert xe.to_id in out.nodes
+            # No further edges from the downstream node.
+            downstream_edges = [e for e in out.edges if e.from_id == xe.to_id]
+            assert len(downstream_edges) == 0
+
+    # Verify hint generation works on the trace output.
+    from mcp_hints import generate_hints
+    trace_payload = {
+        "success": out.success,
+        "stats": out.stats.model_dump(),
+        "edges": [e.model_dump() for e in out.edges],
+        "nodes": {nid: n.model_dump() for nid, n in out.nodes.items()},
+        "seed_ids": out.seed_ids,
+        "direction": out.direction,
+        "edge_types": out.edge_types,
+    }
+    struct, advisories = generate_hints("trace", trace_payload)
+    # Cross-service boundary hints should fire if xs_edges exist.
+    if xs_edges:
+        assert any("cross-service" in a.lower() for a in advisories), (
+            f"expected cross-service advisory, got: {advisories}"
+        )