fix(flow): follow kafka topics + rename --max-hops to --depth

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

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",

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)
 
 

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
 
@@ -620,6 +623,134 @@ 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 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 <topic>`` 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.incoming", "--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}"
+    # 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(
+    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, 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)
+
+    # --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"
+    )
+
+    # --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 <topic>`` 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) -----
 
 

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"):

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
+    <topic>``/``callers <topic>``/``overview <topic>`` 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))