feat(schema): re-identify call_graph endpoints onto can:// ids

Author: rahlk

codeanalyzer/core.py

@@ -21,6 +21,7 @@
 from codeanalyzer.schema.assign_ids import assign_ids
 from codeanalyzer.schema.l1_body import populate_l1_body
 from codeanalyzer.schema.l2_callees import backfill_callees
+from codeanalyzer.schema.call_graph_ids import reidentify_call_graph
 from codeanalyzer.schema.py_schema import PyCallEdge
 from codeanalyzer.semantic_analysis.call_graph import (
     filter_external_edges,
@@ -465,6 +466,7 @@ def analyze(self) -> Analysis:
         populate_l1_body(app)
         if self.analysis_level >= 2:
             backfill_callees(app, sig_to_id)
+        reidentify_call_graph(app, sig_to_id)
 
         # L3/L4 dataflow emission rebuilt on the v2 tree in Stage 3+
 

codeanalyzer/schema/call_graph_ids.py

@@ -0,0 +1,12 @@
+"""Re-identify call-graph edge endpoints onto canonical can:// ids so the JSON
+call_graph agrees with the Neo4j PY_CALLS projection. Declared endpoints map
+through sig_to_id; external/library endpoints keep their dotted signature
+(they have no can:// id)."""
+from __future__ import annotations
+from codeanalyzer.schema.py_schema import PyApplication
+
+
+def reidentify_call_graph(app: PyApplication, sig_to_id: dict) -> None:
+    for edge in app.call_graph or []:
+        edge.source = sig_to_id.get(edge.source, edge.source)
+        edge.target = sig_to_id.get(edge.target, edge.target)

test/test_v2_l2.py

@@ -1,6 +1,7 @@
 from codeanalyzer.schema.l2_callees import backfill_callees
+from codeanalyzer.schema.call_graph_ids import reidentify_call_graph
 from codeanalyzer.schema.py_schema import (
-    PyApplication, PyModule, PyCallable, PyCallsite, BodyNode,
+    PyApplication, PyModule, PyCallable, PyCallsite, BodyNode, PyCallEdge,
 )
 
 def _app_with_one_call(callee_sig):
@@ -25,3 +26,18 @@ def test_unresolved_callsite_leaves_callee_absent():
     app, fn = _app_with_one_call(None)
     backfill_callees(app, {})
     assert fn.body["2:4"].callee is None
+
+def test_call_graph_endpoints_reidentified():
+    edge = PyCallEdge(source="m.f", target="m.g")
+    app = PyApplication(symbol_table={}, call_graph=[edge])
+    reidentify_call_graph(app, {"m.f": "can://python/app/m.py/f()",
+                                "m.g": "can://python/app/m.py/g()"})
+    assert app.call_graph[0].source == "can://python/app/m.py/f()"
+    assert app.call_graph[0].target == "can://python/app/m.py/g()"
+
+def test_call_graph_external_target_unchanged():
+    edge = PyCallEdge(source="m.f", target="requests.get")
+    app = PyApplication(symbol_table={}, call_graph=[edge])
+    reidentify_call_graph(app, {"m.f": "can://python/app/m.py/f()"})
+    assert app.call_graph[0].source == "can://python/app/m.py/f()"
+    assert app.call_graph[0].target == "requests.get"