feat(schema): L2 backfill of body call-node callee (null→id)

Author: rahlk

codeanalyzer/core.py

@@ -20,6 +20,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.py_schema import PyCallEdge
 from codeanalyzer.semantic_analysis.call_graph import (
     filter_external_edges,
@@ -459,8 +460,11 @@ def analyze(self) -> Analysis:
             .build()
         )
 
-        assign_ids(app, self.options.app_name or self.project_dir.name)
+        app_name = self.options.app_name or self.project_dir.name
+        sig_to_id = assign_ids(app, app_name)
         populate_l1_body(app)
+        if self.analysis_level >= 2:
+            backfill_callees(app, sig_to_id)
 
         # L3/L4 dataflow emission rebuilt on the v2 tree in Stage 3+
 

codeanalyzer/schema/l2_callees.py

@@ -0,0 +1,36 @@
+"""L2 refinement: fill each L1 `call` body node's `callee` (null→id) from the
+call site's resolved signature — the one sanctioned value change. A declared
+target becomes its can:// id; an external/library target keeps its dotted
+signature; an unresolved call site leaves `callee` absent."""
+from __future__ import annotations
+from codeanalyzer.schema.py_schema import PyApplication, PyClass, PyCallable
+
+
+def _do_callable(c: PyCallable, sig_to_id: dict) -> None:
+    for cs in c.call_sites or []:
+        if cs.callee_signature is None:
+            continue
+        key = f"{cs.start_line}:{cs.start_column}"
+        node = c.body.get(key)
+        if node is None or node.kind != "call":
+            continue
+        node.callee = sig_to_id.get(cs.callee_signature, cs.callee_signature)
+    for ic in (c.inner_callables or {}).values():
+        _do_callable(ic, sig_to_id)
+    for icl in (c.inner_classes or {}).values():
+        _do_class(icl, sig_to_id)
+
+
+def _do_class(cl: PyClass, sig_to_id: dict) -> None:
+    for m in (cl.methods or {}).values():
+        _do_callable(m, sig_to_id)
+    for ic in (cl.inner_classes or {}).values():
+        _do_class(ic, sig_to_id)
+
+
+def backfill_callees(app: PyApplication, sig_to_id: dict) -> None:
+    for mod in app.symbol_table.values():
+        for fn in (mod.functions or {}).values():
+            _do_callable(fn, sig_to_id)
+        for cl in (mod.classes or {}).values():
+            _do_class(cl, sig_to_id)

test/test_v2_l2.py

@@ -0,0 +1,27 @@
+from codeanalyzer.schema.l2_callees import backfill_callees
+from codeanalyzer.schema.py_schema import (
+    PyApplication, PyModule, PyCallable, PyCallsite, BodyNode,
+)
+
+def _app_with_one_call(callee_sig):
+    cs = PyCallsite(method_name="g", start_line=2, start_column=4, end_line=2,
+                    end_column=7, callee_signature=callee_sig)
+    fn = PyCallable(name="f", path="m.py", signature="m.f", call_sites=[cs],
+                    body={"2:4": BodyNode(kind="call", callee=None)})
+    mod = PyModule(file_path="m.py", module_name="m", functions={"f": fn})
+    return PyApplication(symbol_table={"m.py": mod}), fn
+
+def test_declared_callee_resolves_to_can_id():
+    app, fn = _app_with_one_call("m.g")
+    backfill_callees(app, {"m.g": "can://python/app/m.py/g()"})
+    assert fn.body["2:4"].callee == "can://python/app/m.py/g()"
+
+def test_external_callee_keeps_dotted_signature():
+    app, fn = _app_with_one_call("requests.get")
+    backfill_callees(app, {"m.g": "can://python/app/m.py/g()"})  # requests.get not declared
+    assert fn.body["2:4"].callee == "requests.get"
+
+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