v2 Stage 2: L2 (callee backfill + call_graph re-identity)

codeanalyzer/core.py

@@ -20,6 +20,8 @@
 )
 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,
@@ -459,8 +461,12 @@ 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)
+        reidentify_call_graph(app, sig_to_id)
 
         # L3/L4 dataflow emission rebuilt on the v2 tree in Stage 3+
 

codeanalyzer/neo4j/project.py

@@ -57,15 +57,18 @@ def project(app: PyApplication, app_name: str, sig_to_id: dict) -> GraphRows:
         ["PyApplication"], "name", app_name, {"schema_version": SCHEMA_VERSION}
     )
 
+    # Endpoints listed in app.external_symbols become :PyExternal ghost nodes; the
+    # rest are declared :PySymbol nodes emitted here (keyed by their can:// id,
+    # resolved through ``sig_to_id``). Both the module-body projection (for
+    # PY_EXTENDS / PY_RESOLVES_TO) and the PY_CALLS twin below share this split.
+    externals = app.external_symbols or {}
+
     for file_key, mod in app.symbol_table.items():
         mod_ref = b.node(["PyModule"], "id", mod.id, _module_props(mod, file_key))
         b.edge("PY_HAS_MODULE", app_ref, mod_ref)
-        _project_module_body(b, file_key, mod_ref, mod)
+        _project_module_body(b, file_key, mod_ref, mod, externals, sig_to_id)
 
-    # The aggregated :PY_CALLS twin. Endpoints listed in app.external_symbols become
-    # :PyExternal ghost nodes; the rest are declared :PySymbol nodes already emitted
-    # (keyed by their can:// id, resolved through ``sig_to_id``).
-    externals = app.external_symbols or {}
+    # The aggregated :PY_CALLS twin.
     for e in app.call_graph:
         src = _call_endpoint(b, e.source, externals, sig_to_id)
         tgt = _call_endpoint(b, e.target, externals, sig_to_id)
@@ -175,6 +178,16 @@ def _sym(can_id: str) -> NodeRef:
     return NodeRef("PySymbol", "id", can_id)
 
 
+def _symbol_ref(signature: str, externals: dict, sig_to_id: dict) -> NodeRef:
+    """Resolve a call/inheritance target to the NodeRef under which it was (or
+    will be) emitted: a declared symbol by its can:// id, otherwise a
+    signature-keyed :PySymbol (external ghost)."""
+    can_id = sig_to_id.get(signature)
+    if can_id is not None:
+        return NodeRef("PySymbol", "id", can_id)
+    return NodeRef("PySymbol", "signature", signature)
+
+
 def _call_endpoint(
     b: RowBuilder, signature: str, externals: dict, sig_to_id: dict
 ) -> NodeRef:
@@ -212,12 +225,13 @@ def _call_endpoint(
 
 
 def _project_module_body(
-    b: RowBuilder, file_key: str, mod_ref: NodeRef, mod: PyModule
+    b: RowBuilder, file_key: str, mod_ref: NodeRef, mod: PyModule,
+    externals: dict, sig_to_id: dict,
 ) -> None:
     for fn in (mod.functions or {}).values():
-        _project_callable(b, file_key, mod_ref, "PY_DECLARES", fn)
+        _project_callable(b, file_key, mod_ref, "PY_DECLARES", fn, externals, sig_to_id)
     for cl in (mod.classes or {}).values():
-        _project_class(b, file_key, mod_ref, "PY_DECLARES", cl)
+        _project_class(b, file_key, mod_ref, "PY_DECLARES", cl, externals, sig_to_id)
     for v in mod.variables or []:
         _project_variable(b, file_key, mod_ref, file_key, v)
     _project_imports(b, mod_ref, mod)
@@ -256,26 +270,29 @@ def _project_imports(b: RowBuilder, mod_ref: NodeRef, mod: PyModule) -> None:
 
 
 def _project_class(
-    b: RowBuilder, file_key: str, parent: NodeRef, parent_rel: str, cl: PyClass
+    b: RowBuilder, file_key: str, parent: NodeRef, parent_rel: str, cl: PyClass,
+    externals: dict, sig_to_id: dict,
 ) -> None:
     ref = b.node(
         ["PySymbol", "PyClass"], "id", cl.id, _class_props(cl, file_key)
     )
     b.edge(parent_rel, parent, ref)
 
     for base in cl.base_classes or []:
-        b.edge_to_symbol("PY_EXTENDS", ref, base)
+        if base:
+            b.edge_to_symbol("PY_EXTENDS", ref, _symbol_ref(base, externals, sig_to_id))
 
     for m in (cl.methods or {}).values():
-        _project_callable(b, file_key, ref, "PY_HAS_METHOD", m)
+        _project_callable(b, file_key, ref, "PY_HAS_METHOD", m, externals, sig_to_id)
     for a in (cl.attributes or {}).values():
         _project_attribute(b, file_key, ref, cl.signature, a)
     for ic in (cl.inner_classes or {}).values():
-        _project_class(b, file_key, ref, "PY_DECLARES", ic)
+        _project_class(b, file_key, ref, "PY_DECLARES", ic, externals, sig_to_id)
 
 
 def _project_callable(
-    b: RowBuilder, file_key: str, owner: NodeRef, owner_rel: str, c: PyCallable
+    b: RowBuilder, file_key: str, owner: NodeRef, owner_rel: str, c: PyCallable,
+    externals: dict, sig_to_id: dict,
 ) -> None:
     ref = b.node(
         ["PySymbol", "PyCallable"],
@@ -296,14 +313,17 @@ def _project_callable(
         cs = b.node(["PyCallSite"], "id", cs_id, _call_site_props(s, file_key))
         b.edge("PY_HAS_CALLSITE", ref, cs)
         if s.callee_signature:
-            b.edge_to_symbol("PY_RESOLVES_TO", cs, s.callee_signature)
+            b.edge_to_symbol(
+                "PY_RESOLVES_TO", cs,
+                _symbol_ref(s.callee_signature, externals, sig_to_id),
+            )
 
     for v in c.local_variables or []:
         _project_variable(b, file_key, ref, c.signature, v)
     for ic in (c.inner_callables or {}).values():
-        _project_callable(b, file_key, ref, "PY_DECLARES", ic)
+        _project_callable(b, file_key, ref, "PY_DECLARES", ic, externals, sig_to_id)
     for cl in (c.inner_classes or {}).values():
-        _project_class(b, file_key, ref, "PY_DECLARES", cl)
+        _project_class(b, file_key, ref, "PY_DECLARES", cl, externals, sig_to_id)
 
 
 def _project_attribute(

codeanalyzer/neo4j/rows.py

@@ -110,20 +110,15 @@ def edge(self, type_: str, from_ref: NodeRef, to_ref: NodeRef, props: Optional[P
         self._edges.append(EdgeRow(type_, from_ref, to_ref, dict(props or {})))
 
     def edge_to_symbol(
-        self, type_: str, from_ref: NodeRef, target_signature: str, props: Optional[Props] = None
+        self, type_: str, from_ref: NodeRef, target_ref: NodeRef, props: Optional[Props] = None
     ) -> None:
         """An edge to a ``:PySymbol`` target that may be external/library code not
-        present in the graph. Deferred and kept only if the target signature was
-        actually emitted as a node — so PY_EXTENDS / PY_RESOLVES_TO never dangle (the
-        string fallback lives on the source node's props)."""
-        self._deferred.append(
-            EdgeRow(
-                type_,
-                from_ref,
-                NodeRef("PySymbol", "signature", target_signature),
-                dict(props or {}),
-            )
-        )
+        present in the graph. The target is an already-resolved :class:`NodeRef`
+        (a declared symbol by its can:// id, or a signature-keyed external ghost).
+        Deferred and kept only if that ``(label, value)`` was actually emitted as a
+        node — so PY_EXTENDS / PY_RESOLVES_TO never dangle (the string fallback lives
+        on the source node's props)."""
+        self._deferred.append(EdgeRow(type_, from_ref, target_ref, dict(props or {})))
 
     def has_key(self, label: str, value: str) -> bool:
         """Whether a node with this ``(merge_label, value)`` identity was emitted."""

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)

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/conftest_v2.py

@@ -43,8 +43,12 @@ def assert_conformant(payload: dict, max_level: int) -> None:
         text = mod["source"].encode("utf-8")[lo:hi].decode("utf-8")
         assert text.lstrip().startswith(("def ", "async def ", "@")), f"{c['id']} span mismatch"
         for node in c.get("body", {}).values():
-            if node.get("kind") == "call" and max_level >= 2:
-                assert node.get("callee") is None or isinstance(node["callee"], str)
+            if node.get("kind") == "call" and "callee" in node:
+                assert isinstance(node["callee"], str), "resolved callee must be a string id"
+    if max_level >= 2:
+        for e in app.get("call_graph", []):
+            assert isinstance(e["source"], str), f"call_graph edge source must be string: {e}"
+            assert isinstance(e["target"], str), f"call_graph edge target must be string: {e}"
     for mod, c in _iter_callables(app):
         node_ids = set(c.get("body", {}).keys())
         for lst in ("cfg", "cdg", "ddg", "summary"):

test/test_v2_l2.py

@@ -0,0 +1,43 @@
+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, PyCallEdge,
+)
+
+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
+
+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"

test/test_v2_superset.py

@@ -0,0 +1,54 @@
+import json
+import subprocess
+import sys
+from pathlib import Path
+
+from conftest_v2 import assert_conformant
+
+
+def _run(proj, level):
+    out = subprocess.run(
+        [sys.executable, "-m", "codeanalyzer", "-i", str(proj), "-a", str(level), "--no-venv"],
+        capture_output=True, text=True, check=True,
+    ).stdout
+    return json.loads(out)
+
+
+def _callables(app):
+    def wc(c):
+        yield c
+        for ic in (c.get("inner_callables") or {}).values():
+            yield from wc(ic)
+        for cl in (c.get("inner_classes") or {}).values():
+            yield from wcl(cl)
+
+    def wcl(cl):
+        for m in (cl.get("methods") or {}).values():
+            yield from wc(m)
+        for ic in (cl.get("inner_classes") or {}).values():
+            yield from wcl(ic)
+
+    for mod in app["symbol_table"].values():
+        for fn in (mod.get("functions") or {}).values():
+            yield from wc(fn)
+        for cl in (mod.get("classes") or {}).values():
+            yield from wcl(cl)
+
+
+def test_l1_subset_of_l2(tmp_path):
+    proj = tmp_path / "proj"
+    proj.mkdir()
+    (proj / "m.py").write_text("def g():\n    return 1\ndef f():\n    return g()\n", encoding="utf-8")
+    l1, l2 = _run(proj, 1), _run(proj, 2)
+    assert_conformant(l1, max_level=1)
+    assert_conformant(l2, max_level=2)
+    a1, a2 = l1["application"], l2["application"]
+    # every module + callable id present at L1 is present at L2
+    ids1 = {c["id"] for c in _callables(a1)}
+    ids2 = {c["id"] for c in _callables(a2)}
+    assert ids1 <= ids2, "L2 dropped a callable present at L1"
+    # every L1 body node key is present at L2 (callee may refine null→id)
+    for c1 in _callables(a1):
+        c2 = next(c for c in _callables(a2) if c["id"] == c1["id"])
+        assert set(c1.get("body", {})) <= set(c2.get("body", {})), \
+            f"L2 dropped a body node from {c1['id']}"

test/test_v2_two_projection_agreement.py

@@ -1,6 +1,8 @@
 from codeanalyzer.schema.assign_ids import assign_ids
 from codeanalyzer.neo4j.project import project
-from codeanalyzer.schema.py_schema import PyApplication, PyModule, PyCallable
+from codeanalyzer.schema.py_schema import (
+    PyApplication, PyModule, PyClass, PyCallable, PyCallsite,
+)
 
 
 def test_neo4j_callable_key_equals_json_id():
@@ -13,3 +15,20 @@ def test_neo4j_callable_key_equals_json_id():
     keys = {n.value for n in rows.nodes}
     assert fn.id in keys          # the callable node is keyed by its can:// id
     assert app.symbol_table["m.py"].id in keys
+
+
+def test_py_resolves_to_edge_targets_declared_callee_by_can_id():
+    callee = PyCallable(name="g", path="m.py", signature="m.g", parameters=[])
+    cs = PyCallsite(method_name="g", start_line=2, start_column=4, end_line=2,
+                    end_column=7, callee_signature="m.g")
+    caller = PyCallable(name="f", path="m.py", signature="m.f", parameters=[],
+                        call_sites=[cs])
+    mod = PyModule(file_path="m.py", module_name="m", source="def f():\n    g()\n",
+                   functions={"f": caller, "g": callee})
+    app = PyApplication(symbol_table={"m.py": mod})
+    sig_to_id = assign_ids(app, "myapp")
+    rows = project(app, "myapp", sig_to_id)
+    resolves = [e for e in rows.edges if e.type == "PY_RESOLVES_TO"]
+    # the callsite must resolve to g's can:// id — edge kept, not dropped
+    assert any(e.to_ref.value == sig_to_id["m.g"] for e in resolves), \
+        "PY_RESOLVES_TO must target the declared callee by can:// id"