feat(dataflow): emit L3 body + cfg/cdg/ddg(ssa) onto the v2 tree

Author: rahlk

codeanalyzer/core.py

@@ -468,7 +468,20 @@ def analyze(self) -> Analysis:
             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+
+        # L3: intraprocedural dataflow (CFG/CDG/DDG) emitted onto the v2 tree.
+        if self.analysis_level >= 3:
+            from codeanalyzer.dataflow.builder import (
+                build_function_pdgs,
+                emit_l3_body,
+            )
+            from codeanalyzer.dataflow.syntactic import SyntacticOracle
+
+            infos, _func_asts = build_function_pdgs(
+                app,
+                k=self.options.graph_field_depth,
+                oracle_factory=lambda c: SyntacticOracle(),
+            )
+            emit_l3_body(app, infos, sig_to_id, set(self.options.graphs.split(",")))
 
         # Build the v2 envelope, then persist it (the cache stores the full
         # ``Analysis`` envelope so a reused cache round-trips schema_version).

codeanalyzer/dataflow/builder.py

@@ -192,6 +192,127 @@ def build_function_pdgs(
     return infos, func_asts
 
 
+def emit_l3_body(
+    app: PyApplication,
+    infos: Dict[str, FunctionInfo],
+    sig_to_id: Dict[str, str],
+    graphs: Set[str],
+) -> None:
+    """Project each callable's syntactic PDG onto the v2 tree at L3.
+
+    For every callable that produced a ``FunctionInfo`` in
+    :func:`build_function_pdgs` (syntactic oracle), this writes onto the
+    matching ``PyCallable`` in ``app``'s symbol table:
+
+    * ``body`` — one node per CFG node, keyed by its ordinal id
+      (``<can:// id>@entry``/``@exit`` for the synthetic bookends,
+      ``<can:// id>@line:col`` for real statements). A statement position an
+      L1 pass already materialized as a ``call`` node keeps its ``call`` kind
+      and L2-resolved ``callee``; it is only re-keyed onto its ordinal id (so
+      the edge lists resolve to it and it is not duplicated) and given the
+      byte-offset ``span`` L1 could not compute.
+    * ``cfg`` — one ``CfgEdge`` per CFG edge, endpoints as ordinal ids.
+    * ``cdg`` — the PDG's control-dependence edges.
+    * ``ddg`` — the PDG's syntactic def-use edges, each with ``prov=["ssa"]``
+      (no points-to provenance at L3; that is the L4 delta).
+
+    ``graphs`` scopes the edge lists exactly as the dormant
+    :func:`to_program_graphs` does: ``cfg`` needs ``"cfg"``; ``cdg`` needs
+    ``"pdg"``/``"sdg"``; ``ddg`` needs those or ``"dfg"``. ``body`` is always
+    populated. Callables absent from ``infos`` (unrecovered AST) are skipped.
+    """
+    from codeanalyzer.dataflow.identity import IdentityMap
+    from codeanalyzer.schema.py_schema import (
+        BodyNode,
+        CdgEdge,
+        CfgEdge,
+        DdgEdge,
+        Span,
+        byte_offsets,
+    )
+
+    want_pdg = bool({"pdg", "sdg"} & graphs)
+    want_cfg = "cfg" in graphs
+    want_ddg = want_pdg or "dfg" in graphs
+
+    def _span_of(source: str, node) -> Optional["Span"]:
+        if not source or node.start_line < 1:
+            return None
+        return Span(
+            start=(node.start_line, node.start_column),
+            end=(node.end_line, node.end_column),
+            bytes=byte_offsets(
+                source,
+                node.start_line,
+                node.start_column,
+                node.end_line,
+                node.end_column,
+            ),
+        )
+
+    for module in app.symbol_table.values():
+        source = module.source
+        for pycallable, _chain in _walk_callables(module):
+            info = infos.get(pycallable.signature)
+            if info is None:
+                continue
+            pdg = info.pdg
+            callable_id = sig_to_id.get(pycallable.signature) or pycallable.id
+            im = IdentityMap.for_function(callable_id, pdg)
+
+            for node in pdg.cfg.nodes:
+                ordinal = im.ordinal(node.id)
+                if node.id == pdg.cfg.entry_id:
+                    pycallable.body[ordinal] = BodyNode(kind="entry")
+                    continue
+                if node.id == pdg.cfg.exit_id:
+                    pycallable.body[ordinal] = BodyNode(kind="exit")
+                    continue
+                span = _span_of(source, node)
+                # An L1 `call` node was keyed by its "line:col"; if this CFG node
+                # sits at the same position, keep that node's `call` kind and
+                # resolved `callee` and merely re-key it onto the ordinal id
+                # (dedup + endpoint resolution), filling any missing span.
+                existing = pycallable.body.get(ordinal)
+                if existing is None:
+                    existing = pycallable.body.pop(
+                        f"{node.start_line}:{node.start_column}", None
+                    )
+                if existing is not None:
+                    if existing.span is None and span is not None:
+                        existing.span = span
+                    pycallable.body[ordinal] = existing
+                    continue
+                pycallable.body[ordinal] = BodyNode(kind=node.kind, span=span)
+
+            if want_cfg:
+                pycallable.cfg = [
+                    CfgEdge(
+                        src=im.ordinal(e.source),
+                        dst=im.ordinal(e.target),
+                        kind=e.kind,
+                    )
+                    for e in pdg.cfg.edges
+                ]
+            if want_pdg:
+                pycallable.cdg = [
+                    CdgEdge(src=im.ordinal(e.source), dst=im.ordinal(e.target))
+                    for e in pdg.edges
+                    if e.type == "CDG"
+                ]
+            if want_ddg:
+                pycallable.ddg = [
+                    DdgEdge(
+                        src=im.ordinal(e.source),
+                        dst=im.ordinal(e.target),
+                        var=e.var,
+                        prov=["ssa"],
+                    )
+                    for e in pdg.edges
+                    if e.type == "DDG"
+                ]
+
+
 def build_program_graphs(
     app: PyApplication,
     k: int = DEFAULT_K_LIMIT,

test/test_v2_l3.py

@@ -1,9 +1,10 @@
 import textwrap
 from pathlib import Path
 
-from codeanalyzer.dataflow.builder import build_function_pdgs
+from codeanalyzer.dataflow.builder import build_function_pdgs, emit_l3_body
 from codeanalyzer.dataflow.identity import IdentityMap
 from codeanalyzer.dataflow.syntactic import SyntacticOracle
+from codeanalyzer.schema.assign_ids import assign_ids
 from codeanalyzer.schema.py_schema import PyApplication
 from codeanalyzer.syntactic_analysis.symbol_table_builder import SymbolTableBuilder
 
@@ -37,6 +38,29 @@ def test_syntactic_oracle_only_identity_aliases():
     assert o.may_alias("a", "b") is False
 
 
+def test_emit_l3_populates_body_and_cfg(tmp_path: Path):
+    f = tmp_path / "m.py"
+    f.write_text("def f(a):\n    b = a\n    return b\n", encoding="utf-8")
+    mod = SymbolTableBuilder(tmp_path, None).build_pymodule_from_file(f)
+    app = PyApplication(symbol_table={"m.py": mod})
+    sig_to_id = assign_ids(app, "app")
+    infos, _func_asts = build_function_pdgs(
+        app, k=3, oracle_factory=lambda c: SyntacticOracle()
+    )
+    emit_l3_body(app, infos, sig_to_id, graphs={"cfg", "dfg", "pdg"})
+    fn = next(iter(mod.functions.values()))
+    assert any(k.endswith("@entry") for k in fn.body)
+    assert any(k.endswith("@exit") for k in fn.body)
+    assert len(fn.cfg) > 0
+    # every cfg endpoint resolves to a body node id
+    body_ids = set(fn.body)
+    for e in fn.cfg:
+        assert e.src in body_ids and e.dst in body_ids
+    # ddg (if any) carries ssa provenance
+    for e in fn.ddg:
+        assert e.prov == ["ssa"]
+
+
 def test_build_function_pdgs_returns_pdg_per_callable(tmp_path: Path):
     f = tmp_path / "m.py"
     f.write_text(textwrap.dedent("def f(a):\n    b = a\n    return b\n"), encoding="utf-8")