From ce152e1f2abd07bfc2f4784def9a03d374dc77f9 Mon Sep 17 00:00:00 2001 From: Dmitry Teryaev Date: Tue, 7 Jul 2026 00:22:35 +0300 Subject: [PATCH 1/3] fix(resolve): match r.topic so kafka_topic Routes resolve by name _resolve_route_candidates matched only on path/path_template, so kafka_topic Routes (name in `topic`, `path=''`) were unresolvable by name. `jrag flow ` / `callers ` / `overview ` returned 'none' even though the route + EXPOSES edge existed, while trace_request_flow already supported kafka end-to-end (async inbound via ASYNC_CALLS/topic + outbound via EXPOSES). Add r.topic exact + prefix queries mirroring _resolve_producer_candidates (new route_topic / route_topic_prefix reasons). _drop_route_mirrors already collapses the no-EXPOSES producer phantom onto the @KafkaListener server route (kafka fqn='' groups them), so a producer+consumer topic resolves to the single server route. Co-Authored-By: Claude --- java_ontology.py | 4 +++ resolve_service.py | 24 ++++++++++++++++++ tests/test_mcp_v2.py | 29 +++++++++++++++++++++ tests/test_resolve_service.py | 47 +++++++++++++++++++++++++++++++++++ 4 files changed, 104 insertions(+) diff --git a/java_ontology.py b/java_ontology.py index c7dffa84..e42588f0 100644 --- a/java_ontology.py +++ b/java_ontology.py @@ -90,6 +90,8 @@ "short_name", "route_template", "route_method_path", + "route_topic", + "route_topic_prefix", "client_target", "client_target_path", "client_name", @@ -428,6 +430,8 @@ class EdgeSpec: "short_name", "route_template", "route_method_path", + "route_topic", + "route_topic_prefix", "client_target", "client_target_path", "client_name", diff --git a/resolve_service.py b/resolve_service.py index 75b40556..c0f61ba2 100644 --- a/resolve_service.py +++ b/resolve_service.py @@ -43,11 +43,13 @@ "producer_topic_prefix": 1, "fqn_suffix": 2, "route_template": 2, + "route_topic": 2, "client_fqn": 2, "short_name": 3, "client_target": 3, "client_name": 3, "producer_topic": 3, + "route_topic_prefix": 3, } _SYMBOL_RESOLVE_RETURN = ( @@ -295,6 +297,28 @@ def _resolve_route_candidates( path_val = str(row.get("path_template") or row.get("path") or "") out.append((_node_ref_from_row("route", row), "route_template", len(path_val))) + # Kafka/topic routes carry their name in ``topic`` (``path``/``path_template`` + # are empty), so path-based matching above cannot reach them. Match on + # ``r.topic`` the same way ``_resolve_producer_candidates`` matches + # ``p.topic`` — this lets ``flow``/``callers``/``overview`` resolve a + # ``kafka_topic`` Route by topic name. ``_drop_route_mirrors`` below then + # discards the no-EXPOSES producer phantom in favour of the server route. + rows = g._rows( # noqa: SLF001 + f"MATCH (r:Route) WHERE r.topic = $topic{scope} RETURN {_ROUTE_RESOLVE_RETURN} LIMIT $lim", + {"topic": identifier, "lim": lim, **scope_params}, + ) + for row in rows: + out.append((_node_ref_from_row("route", row), "route_topic", len(identifier))) + + if not identifier.startswith("/"): + rows = g._rows( # noqa: SLF001 + f"MATCH (r:Route) WHERE r.topic STARTS WITH $topic{scope} " + f"RETURN {_ROUTE_RESOLVE_RETURN} LIMIT $lim", + {"topic": identifier, "lim": lim, **scope_params}, + ) + for row in rows: + out.append((_node_ref_from_row("route", row), "route_topic_prefix", len(identifier))) + return _drop_route_mirrors(g, out) diff --git a/tests/test_mcp_v2.py b/tests/test_mcp_v2.py index 57e08c69..080fb8d7 100644 --- a/tests/test_mcp_v2.py +++ b/tests/test_mcp_v2.py @@ -1364,6 +1364,27 @@ def test_resolve_every_reason_in_closed_set_appears() -> None: "end_line": 1, "resolved": True, } + # kafka_topic Route: name lives in ``topic`` (path empty) — exercises the + # r.topic resolve branch (route_topic / route_topic_prefix reasons). + kafka_route_row = { + "id": "route:reasontopic000", + "kind": "kafka_topic", + "framework": "kafka", + "method": "", + "path": "", + "path_template": "", + "path_regex": "", + "topic": "reason.created", + "broker": "", + "feign_name": "", + "feign_url": "", + "microservice": "svc", + "module": "mod", + "filename": "K.java", + "start_line": 1, + "end_line": 1, + "resolved": True, + } client_row = { "id": "client:reason", "client_kind": "feign_method", @@ -1415,6 +1436,10 @@ def _rows(self, query: str, params: dict | None = None) -> list: return [route_row] if "r.path = $path OR r.path_template = $path" in query: return [route_row] + if "r.topic STARTS WITH $topic" in query: + return [kafka_route_row] + if "WHERE r.topic = $topic" in query: + return [kafka_route_row] if "WHERE c.id = $id" in query: return [client_row] if "STARTS WITH $path" in query: @@ -1443,6 +1468,10 @@ def _rows(self, query: str, params: dict | None = None) -> list: seen.add(reason) for _node, reason, _spec in _resolve_route_candidates(g, "/reason"): seen.add(reason) + # Topic resolution: exact ``r.topic`` (route_topic) + prefix + # ``r.topic STARTS WITH`` (route_topic_prefix, since no leading '/'). + for _node, reason, _spec in _resolve_route_candidates(g, "reason.created"): + seen.add(reason) for _node, reason, _spec in _resolve_client_candidates(g, "client:reason"): seen.add(reason) for _node, reason, _spec in _resolve_client_candidates(g, "reasonsvc"): diff --git a/tests/test_resolve_service.py b/tests/test_resolve_service.py index 399c2e9a..55bcff4e 100644 --- a/tests/test_resolve_service.py +++ b/tests/test_resolve_service.py @@ -146,6 +146,53 @@ def test_resolve_service_producer_topic_prefix(ladybug_db_path: Path) -> None: assert result.message is not None +def test_resolve_service_route_kafka_topic(ladybug_db_path: Path) -> None: + """A Kafka topic name resolves to its Route via ``r.topic``. + + Regression: ``_resolve_route_candidates`` matched only on + ``path``/``path_template``, so ``kafka_topic`` Routes (which carry their + name in ``topic`` with ``path=''``) were unresolvable — ``jrag flow + ``/``callers ``/``overview `` could not follow kafka + even when the route + EXPOSES edge existed. ``banking.chat.compliance.review`` + is consumed by ``ComplianceReviewListener`` (consumer-only, no producer + phantom), so it resolves to exactly one Route. + """ + g = LadybugGraph.get(str(ladybug_db_path)) + result = resolve_v2("banking.chat.compliance.review", hint_kind="route", graph=g) + + assert isinstance(result, ResolveOutput) + assert result.success is True + assert result.resolved_identifier == "banking.chat.compliance.review" + assert result.status == "one", ( + f"topic should resolve to one Route, got status={result.status!r}: {result.message}" + ) + assert result.node is not None + assert result.node.kind == "route", f"expected a Route node, got {result.node}" + + +def test_resolve_service_route_kafka_topic_drops_producer_mirror(ladybug_db_path: Path) -> None: + """A topic with both a producer (phantom, no EXPOSES) and a consumer + (server Route, EXPOSES) resolves to the single server Route. + + ``banking.chat.incoming`` is produced by ``FollowUpKafkaPublisher`` (phantom + kafka_topic Route, ``microservice=''``, no EXPOSES) and consumed by + ``ChatKafkaListener`` (server Route with EXPOSES). Both carry the same topic + so topic-matching surfaces both; ``_drop_route_mirrors`` must collapse them + to the one exposed server route (status ``one``), not report ``many``. + """ + g = LadybugGraph.get(str(ladybug_db_path)) + result = resolve_v2("banking.chat.incoming", hint_kind="route", graph=g) + + assert isinstance(result, ResolveOutput) + assert result.success is True + assert result.status == "one", ( + f"producer+consumer topic should collapse to one server Route via mirror-drop, " + f"got status={result.status!r}: {result.message}" + ) + assert result.node is not None + assert result.node.kind == "route", f"expected a Route node, got {result.node}" + + def test_resolve_service_hint_kind_filters(ladybug_db_path: Path) -> None: """hint_kind narrows the search space (route hint won't match symbol-only ids).""" g = LadybugGraph.get(str(ladybug_db_path)) From d62f04617d2298fc88b67617cb76ddbaf109d3c0 Mon Sep 17 00:00:00 2001 From: Dmitry Teryaev Date: Tue, 7 Jul 2026 00:22:44 +0300 Subject: [PATCH 2/3] feat(flow): rename --max-hops to --depth and accept topic names Rename flow's hop-count flag to --depth so it matches callers/callees/ impact/decompose. --max-hops is kept as a hidden back-compat alias (argparse.SUPPRESS, same dest=depth). The query positional now also accepts a Kafka topic name, made reachable by the resolver fix. Co-Authored-By: Claude --- java_codebase_rag/jrag.py | 24 ++++++++++-- tests/test_jrag_traversal_direct.py | 61 +++++++++++++++++++++++++++++ 2 files changed, 81 insertions(+), 4 deletions(-) diff --git a/java_codebase_rag/jrag.py b/java_codebase_rag/jrag.py index 60b80206..ed95426d 100644 --- a/java_codebase_rag/jrag.py +++ b/java_codebase_rag/jrag.py @@ -812,11 +812,27 @@ def _core_parser() -> argparse.ArgumentParser: "CALLS hops from the route handler. Intra-service is an INDEX-TIME data " "property: CALLS edges are intra-codebase by construction, and the query " "carries no microservice predicate, so the result reflects whatever the " - "fixture indexed (no query-time constraint). --max-hops clamped to 1..8." + "fixture indexed (no query-time constraint). --depth clamped to 1..8." ), ) - flow.add_argument("query", help="Route path (e.g. '/chat/assign'). Resolved with hint_kind=route.") - flow.add_argument("--max-hops", type=int, default=5, dest="max_hops", help="Max CALLS hops (clamped 1..8, default 5).") + flow.add_argument( + "query", + help=( + "Route path (e.g. '/chat/assign') or Kafka topic name (e.g. " + "'banking.chat.compliance.review'). Resolved with hint_kind=route; " + "kafka_topic Routes match on topic." + ), + ) + # Primary flag is --depth (consistent with callers/callees/impact/decompose). + # --max-hops is kept as a hidden back-compat alias (same dest). + flow.add_argument( + "--depth", type=int, default=5, dest="depth", + help="Max CALLS hops (clamped 1..8, default 5).", + ) + flow.add_argument( + "--max-hops", type=int, dest="depth", + default=argparse.SUPPRESS, help=argparse.SUPPRESS, + ) flow.set_defaults(handler=_cmd_flow) # ---- Compose traversals + file inspection (PR-JRAG-3b) ---- @@ -3001,7 +3017,7 @@ def _cmd_flow(args: argparse.Namespace) -> int: args, reason="trace_request_flow carries no microservice predicate; intra-codebase is an index-time data property" ) - max_hops = max(1, min(8, getattr(args, "max_hops", 5))) + max_hops = max(1, min(8, getattr(args, "depth", 5))) flow_data = graph.trace_request_flow(entry_route_id=node.id, max_hops=max_hops) root_id = node.id diff --git a/tests/test_jrag_traversal_direct.py b/tests/test_jrag_traversal_direct.py index f91fb5fa..4b776615 100644 --- a/tests/test_jrag_traversal_direct.py +++ b/tests/test_jrag_traversal_direct.py @@ -620,6 +620,67 @@ def test_flow_outbound_intra_service_on_fixture( ) +def test_flow_follows_kafka_topic_on_fixture( + corpus_root: Path, ladybug_db_path: Path +) -> None: + """flow resolves a Kafka topic name to its Route and follows it. + + Regression: ``_resolve_route_candidates`` matched only on path, so a + ``kafka_topic`` Route (name in ``topic``, ``path=''``) was unresolvable and + ``jrag flow `` returned 'none' even though the route + EXPOSES edge + existed. ``banking.chat.compliance.review`` is consumed by + ``ComplianceReviewListener`` (``@CodebaseAsyncRoute(topic=...)``); flow must + now resolve it to a Route root and surface the listener's outbound CALLS. + """ + env = _env_for(corpus_root, ladybug_db_path) + proc = _run_jrag( + ["flow", "banking.chat.compliance.review", "--format", "json"], env=env + ) + assert proc.returncode == 0, ( + f"flow on kafka topic failed: rc={proc.returncode}\nstdout={proc.stdout}\nstderr={proc.stderr}" + ) + payload = json.loads(proc.stdout) + assert payload["status"] == "ok", f"expected ok, got {payload}" + assert payload.get("root"), "expected root id (the kafka_topic Route)" + root_node = payload.get("nodes", {}).get(payload["root"], {}) + assert root_node.get("kind") == "route", f"expected route root, got {root_node}" + + +def test_flow_depth_flag_and_max_hops_alias( + corpus_root: Path, ladybug_db_path: Path +) -> None: + """flow uses --depth (consistent with callers/callees/impact/decompose). + + --max-hops remains as a hidden back-compat alias (same dest). Both must be + accepted and produce the same shape as the default-depth flow. + """ + env = _env_for(corpus_root, ladybug_db_path) + + # --depth is the primary flag now. + proc_depth = _run_jrag( + ["flow", "/chat/assign", "--depth", "2", "--format", "json"], env=env + ) + assert proc_depth.returncode == 0, ( + f"flow --depth failed: rc={proc_depth.returncode}\nstdout={proc_depth.stdout}" + ) + payload_depth = json.loads(proc_depth.stdout) + assert payload_depth["status"] == "ok", f"expected ok, got {payload_depth}" + + # --max-hops is the hidden alias; must still be accepted (back-compat). + proc_alias = _run_jrag( + ["flow", "/chat/assign", "--max-hops", "2", "--format", "json"], env=env + ) + assert proc_alias.returncode == 0, ( + f"flow --max-hops alias failed: rc={proc_alias.returncode}\nstdout={proc_alias.stdout}" + ) + payload_alias = json.loads(proc_alias.stdout) + assert payload_alias["status"] == "ok", f"expected ok, got {payload_alias}" + # Same dest -> same traversal (depth 2 in both cases). + assert payload_alias.get("root") == payload_depth.get("root"), ( + "expected identical root id for --depth and --max-hops" + ) + + # ----- Test 16: traversal resolve-ambiguous stops (no auto-pick) ----- From 5a0da65b23b5911a24ff1cbdeeeb92ec490e33bd Mon Sep 17 00:00:00 2001 From: Dmitry Teryaev Date: Tue, 7 Jul 2026 00:45:37 +0300 Subject: [PATCH 3/3] test(flow): assert kafka follow edges, --depth effect, callers disambiguation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address code-review feedback (test-coverage gaps; no behavior change): - test_flow_follows_kafka_topic_on_fixture now uses banking.chat.incoming (producer + listener) and asserts a non-empty follow graph WITH an ASYNC_CALLS edge — the previous compliance.review target has an empty listener body, so it only proved resolution, not that flow follows kafka. - test_flow_depth_flag_and_max_hops_alias now also asserts --depth 1 yields strictly fewer edges than --depth 5 (proves the flag affects traversal, not just parses). - New test_callers_topic_disambiguates_with_kind: documents that a dual-sided topic (producer + route) resolves ambiguously without --kind, and that --kind route is the working disambiguation for callers (callers rejects a producer root; use the `producers` command for that view). Co-Authored-By: Claude --- tests/test_jrag_traversal_direct.py | 88 ++++++++++++++++++++++++++--- 1 file changed, 79 insertions(+), 9 deletions(-) diff --git a/tests/test_jrag_traversal_direct.py b/tests/test_jrag_traversal_direct.py index 4b776615..9e1104d6 100644 --- a/tests/test_jrag_traversal_direct.py +++ b/tests/test_jrag_traversal_direct.py @@ -22,8 +22,11 @@ 13. test_impact_service_post_filter_emits_warning 14. test_decompose_renders_role_waterfall 15. test_flow_outbound_intra_service_on_fixture -16. test_traversal_resolve_ambiguous_stops -17. test_traversal_rejects_offset +16. test_flow_follows_kafka_topic_on_fixture +17. test_flow_depth_flag_and_max_hops_alias +18. test_callers_topic_disambiguates_with_kind +19. test_traversal_resolve_ambiguous_stops +20. test_traversal_rejects_offset """ from __future__ import annotations @@ -623,18 +626,22 @@ def test_flow_outbound_intra_service_on_fixture( def test_flow_follows_kafka_topic_on_fixture( corpus_root: Path, ladybug_db_path: Path ) -> None: - """flow resolves a Kafka topic name to its Route and follows it. + """flow resolves a Kafka topic to its Route AND follows it (async edges). Regression: ``_resolve_route_candidates`` matched only on path, so a ``kafka_topic`` Route (name in ``topic``, ``path=''``) was unresolvable and - ``jrag flow `` returned 'none' even though the route + EXPOSES edge - existed. ``banking.chat.compliance.review`` is consumed by - ``ComplianceReviewListener`` (``@CodebaseAsyncRoute(topic=...)``); flow must - now resolve it to a Route root and surface the listener's outbound CALLS. + ``jrag flow `` returned 'none'. Resolution alone is not enough — this + test also asserts the follow graph is non-empty and carries an + ``ASYNC_CALLS`` edge, proving the kafka-specific inbound arm (topic-matched + Producer) actually fires once the route resolves. + + ``banking.chat.incoming`` is the canonical dual-sided topic: produced by + ``FollowUpKafkaPublisher`` (inbound ``ASYNC_CALLS``) and consumed by + ``ChatKafkaListener`` (outbound ``CALLS`` to ``orchestrationService.handle``). """ env = _env_for(corpus_root, ladybug_db_path) proc = _run_jrag( - ["flow", "banking.chat.compliance.review", "--format", "json"], env=env + ["flow", "banking.chat.incoming", "--format", "json"], env=env ) assert proc.returncode == 0, ( f"flow on kafka topic failed: rc={proc.returncode}\nstdout={proc.stdout}\nstderr={proc.stderr}" @@ -644,6 +651,13 @@ def test_flow_follows_kafka_topic_on_fixture( assert payload.get("root"), "expected root id (the kafka_topic Route)" root_node = payload.get("nodes", {}).get(payload["root"], {}) assert root_node.get("kind") == "route", f"expected route root, got {root_node}" + # The follow graph must be non-empty AND carry a kafka async inbound edge — + # this is the part that proves "flow follows kafka", not just "resolves it". + edges = payload.get("edges", []) + assert edges, f"expected non-empty follow graph for a kafka topic, got edges={edges}" + assert any(e.get("edge_type") == "ASYNC_CALLS" for e in edges), ( + f"expected an ASYNC_CALLS (topic-matched Producer) inbound edge, got {edges}" + ) def test_flow_depth_flag_and_max_hops_alias( @@ -652,7 +666,8 @@ def test_flow_depth_flag_and_max_hops_alias( """flow uses --depth (consistent with callers/callees/impact/decompose). --max-hops remains as a hidden back-compat alias (same dest). Both must be - accepted and produce the same shape as the default-depth flow. + accepted, produce the same traversal, and --depth must actually change the + traversal depth (more hops => strictly more CALLS edges). """ env = _env_for(corpus_root, ladybug_db_path) @@ -680,6 +695,61 @@ def test_flow_depth_flag_and_max_hops_alias( "expected identical root id for --depth and --max-hops" ) + # --depth must actually affect the traversal: depth 1 yields strictly fewer + # outbound CALLS edges than depth 5 on this fixture (~3 vs ~20). + proc_shallow = _run_jrag( + ["flow", "/chat/assign", "--depth", "1", "--format", "json"], env=env + ) + proc_deep = _run_jrag( + ["flow", "/chat/assign", "--depth", "5", "--format", "json"], env=env + ) + shallow = json.loads(proc_shallow.stdout) + deep = json.loads(proc_deep.stdout) + assert len(shallow.get("edges", [])) < len(deep.get("edges", [])), ( + f"--depth should change traversal size; depth=1 -> {len(shallow.get('edges', []))}, " + f"depth=5 -> {len(deep.get('edges', []))}" + ) + + +def test_callers_topic_disambiguates_with_kind( + corpus_root: Path, ladybug_db_path: Path +) -> None: + """A dual-sided topic resolves to BOTH a Producer and a Route -> ambiguous + without --kind; --kind route collapses to one and runs find_route_callers. + + Behavior note (from enabling r.topic resolution): ``callers `` with no + --kind now searches all kinds, and a topic with both a Producer and a server + Route is genuinely ambiguous (previously only the Producer matched, which + then errored because callers only accepts Symbol/Route roots — jrag.py:2371). + ``--kind route`` is the disambiguation that yields a working callers run. + (--kind producer resolves to the Producer but callers rejects a Producer + root; use the ``producers`` command for the producer view.) + """ + env = _env_for(corpus_root, ladybug_db_path) + + # No --kind: Producer + Route both match -> ambiguous (no traversal). + proc_both = _run_jrag( + ["callers", "banking.chat.incoming", "--format", "json"], env=env + ) + payload_both = json.loads(proc_both.stdout) + assert payload_both["status"] == "ambiguous", ( + f"expected ambiguous for dual-sided topic without --kind, got {payload_both.get('status')}" + ) + assert len(payload_both.get("candidates", [])) >= 2, ( + f"expected >=2 candidates (producer + route), got {payload_both.get('candidates')}" + ) + + # --kind route: resolves to the single server Route and runs find_route_callers. + proc_route = _run_jrag( + ["callers", "banking.chat.incoming", "--kind", "route", "--format", "json"], + env=env, + ) + payload_route = json.loads(proc_route.stdout) + assert payload_route["status"] == "ok", ( + f"expected --kind route to resolve and run, got {payload_route.get('status')}" + ) + assert payload_route.get("root"), "expected a resolved root for --kind route" + # ----- Test 16: traversal resolve-ambiguous stops (no auto-pick) -----