v2 Stage 3: L3 intraprocedural CFG/CDG/DDG

codeanalyzer/__main__.py

@@ -115,7 +115,7 @@ def main(
             "-a",
             "--analysis-level",
             help="Analysis depth: 1=symbol table+Jedi call graph, 2=+PyCG call "
-            "graph, 3=+native dataflow graphs (CFG/PDG/SDG).",
+            "graph, 3=+native intraprocedural dataflow (CFG/PDG).",
             min=1,
             max=3,
         ),
@@ -125,10 +125,10 @@ def main(
         typer.Option(
             "--graphs",
             help="Level 3 only: comma-separated program-graph sections to emit "
-            "(cfg, dfg, pdg, sdg). Default: all. `dfg` emits the PDG's data "
-            "edges only; `sdg` implies the dependence edges it stitches.",
+            "(cfg, dfg, pdg, sdg). Default: cfg,dfg,pdg. `dfg` emits the PDG's data "
+            "edges only; `sdg` requires -a 4 (not yet available).",
         ),
-    ] = "cfg,dfg,pdg,sdg",
+    ] = "cfg,dfg,pdg",
     graph_field_depth: Annotated[
         int,
         typer.Option(
@@ -277,7 +277,10 @@ def main(
     if not selected_graphs:
         logger.error("--graphs requires at least one of: " + ", ".join(VALID_GRAPHS))
         raise typer.Exit(code=2)
-    if analysis_level < 3 and graphs != "cfg,dfg,pdg,sdg":
+    if "sdg" in selected_graphs:
+        logger.error("--graphs sdg requires -a 4 (interprocedural SDG); not available yet.")
+        raise typer.Exit(code=2)
+    if analysis_level < 3 and graphs != "cfg,dfg,pdg":
         logger.error("--graphs is a level-3 option; pass -a 3 to emit program graphs.")
         raise typer.Exit(code=2)
     if analysis_level < 3 and graph_field_depth != 3:

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).
@@ -507,6 +520,16 @@ def _load_pyapplication_from_cache(self, cache_file: Path) -> Optional[Analysis]
         if getattr(cached, "schema_version", None) != "2.0.0":
             logger.info("stale/incompatible analysis cache (schema_version) — rebuilding")
             return None
+        # The cache keys only on file hash/mtime/size, not on level, so a cache
+        # built at a different analysis_level would leak higher-level body/edge
+        # content (or omit content when the cached level is lower). Reject the
+        # mismatch and force a full rebuild at the requested level.
+        if cached.max_level != self.analysis_level:
+            logger.info(
+                f"cache built at level {cached.max_level} != requested "
+                f"{self.analysis_level} — rebuilding"
+            )
+            return None
         return cached
 
     def _save_analysis_cache(self, analysis: Analysis, cache_file: Path) -> None:

codeanalyzer/dataflow/builder.py

@@ -36,7 +36,7 @@
 
 import ast
 from pathlib import Path
-from typing import Dict, List, Optional, Set, Tuple
+from typing import Callable, Dict, List, Optional, Set, Tuple
 
 from codeanalyzer.dataflow.access_paths import _PathExtractor, _calls_in
 from codeanalyzer.dataflow.alias import TypeBasedAliasOracle
@@ -127,14 +127,24 @@ def _match_args(
     return tuple(pairs)
 
 
-def build_program_graphs(
+def build_function_pdgs(
     app: PyApplication,
     k: int = DEFAULT_K_LIMIT,
-) -> ProgramGraphsIR:
-    """Build CFG/PDG per callable and the whole-program SDG."""
-    class_idx = _class_index(app)
-    callable_idx = _callable_index(app)
-
+    *,
+    oracle_factory: Callable[[PyCallable], object],
+) -> Tuple[Dict[str, FunctionInfo], Dict[str, ast.AST]]:
+    """Intraprocedural phase only: one ``FunctionInfo`` (CFG → PDG) per
+    callable, keyed by signature, with no SDG/summary/callsite work.
+
+    ``oracle_factory(pycallable)`` supplies the may-alias oracle per callable —
+    ``TypeBasedAliasOracle`` for the L4 path, ``SyntacticOracle`` for L3.
+
+    Returns ``(infos, func_asts)`` rather than bare PDGs so that the L4
+    orchestrator (:func:`build_program_graphs`) still has both the
+    ``FunctionInfo`` records its callsite/summary/SDG phases mutate and the
+    matched def nodes its Phase 2 reads. L3 callers just read ``info.pdg`` per
+    signature and ignore ``func_asts``.
+    """
     infos: Dict[str, FunctionInfo] = {}
     func_asts: Dict[str, ast.AST] = {}
 
@@ -166,7 +176,7 @@ def build_program_graphs(
                 if enclosing_ast is not None:
                     enclosing_locals |= _locals_of(enclosing_ast)
 
-            oracle = TypeBasedAliasOracle(_base_types(pycallable))
+            oracle = oracle_factory(pycallable)
             pdg = build_pdg(
                 func,
                 enclosing_locals=enclosing_locals,
@@ -179,6 +189,137 @@ def build_program_graphs(
             )
             func_asts[pycallable.signature] = func
 
+    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 LOCAL id (``"@entry"``/
+      ``"@exit"`` for the synthetic bookends, ``"line:col"`` for real
+      statements — the same key format L1 uses). A statement position an L1
+      pass already materialized as a ``call`` node lands on the SAME local key,
+      so it keeps its ``call`` kind and L2-resolved ``callee`` in place (no
+      re-keying, no duplication) and is only given the byte-offset ``span`` L1
+      could not compute.
+    * ``cfg`` — one ``CfgEdge`` per CFG edge, endpoints as local 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:
+                local = im.local(node.id)
+                if node.id == pdg.cfg.entry_id:
+                    pycallable.body[local] = BodyNode(kind="entry")
+                    continue
+                if node.id == pdg.cfg.exit_id:
+                    pycallable.body[local] = BodyNode(kind="exit")
+                    continue
+                span = _span_of(source, node)
+                # An L1 `call` node was keyed by its LOCAL "line:col"; this CFG
+                # node at the same position lands on the SAME key, so keep the
+                # node's `call` kind and L2-resolved `callee` in place and just
+                # fill any missing span — never re-key or duplicate it.
+                existing = pycallable.body.get(local)
+                if existing is not None:
+                    if existing.span is None and span is not None:
+                        existing.span = span
+                    continue
+                pycallable.body[local] = BodyNode(kind=node.kind, span=span)
+
+            if want_cfg:
+                pycallable.cfg = [
+                    CfgEdge(
+                        src=im.local(e.source),
+                        dst=im.local(e.target),
+                        kind=e.kind,
+                    )
+                    for e in pdg.cfg.edges
+                ]
+            if want_pdg:
+                pycallable.cdg = [
+                    CdgEdge(src=im.local(e.source), dst=im.local(e.target))
+                    for e in pdg.edges
+                    if e.type == "CDG"
+                ]
+            if want_ddg:
+                pycallable.ddg = [
+                    DdgEdge(
+                        src=im.local(e.source),
+                        dst=im.local(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,
+) -> ProgramGraphsIR:
+    """Build CFG/PDG per callable and the whole-program SDG."""
+    class_idx = _class_index(app)
+    callable_idx = _callable_index(app)
+
+    infos, func_asts = build_function_pdgs(
+        app, k, oracle_factory=lambda c: TypeBasedAliasOracle(_base_types(c))
+    )
+
     # Callsites and nested defs, now that every signature is known.
     for sig, info in infos.items():
         pycallable = callable_idx[sig]

codeanalyzer/dataflow/identity.py

@@ -0,0 +1,48 @@
+"""Bijection between internal IR node ids (ints, per function) and their
+canonical ids.
+
+Two forms per node:
+
+* **local** — the intra-callable id used as the ``body`` map key and as every
+  ``cfg``/``cdg``/``ddg`` edge endpoint: ``"@entry"``/``"@exit"`` for the
+  synthetic CFG bookends, ``"line:col"`` for real statements. This matches the
+  key format L1 already uses for ``call`` nodes (see ``schema/l1_body.py``), so
+  an L1 body node and its coinciding CFG node land on the same key and L1 ⊆ L3
+  holds.
+* **global** — ``"<callable can:// id>@<local>"``, the fully addressable id for
+  cross-callable references and the Neo4j PyCFGNode keys (a later task).
+"""
+from __future__ import annotations
+from typing import Dict, Iterable
+
+
+class IdentityMap:
+    def __init__(self, callable_id: str, id_to_local: Dict[int, str]):
+        self._callable_id = callable_id
+        self._map = id_to_local
+
+    @classmethod
+    def for_function(cls, callable_id: str, pdg) -> "IdentityMap":
+        cfg = pdg.cfg
+        m: Dict[int, str] = {}
+        for n in cfg.nodes:
+            if n.id == cfg.entry_id:
+                m[n.id] = "@entry"
+            elif n.id == cfg.exit_id:
+                m[n.id] = "@exit"
+            else:
+                m[n.id] = f"{n.start_line}:{n.start_column}"
+        return cls(callable_id, m)
+
+    def local(self, node_id: int) -> str:
+        """Intra-callable id: ``"@entry"``/``"@exit"`` or ``"line:col"``."""
+        return self._map[node_id]
+
+    def global_id(self, node_id: int) -> str:
+        """Fully addressable id: ``"<callable-id>@<local>"``."""
+        loc = self._map[node_id]
+        # local statements are "line:col"; bookends already carry the leading "@"
+        return f"{self._callable_id}{loc}" if loc.startswith("@") else f"{self._callable_id}@{loc}"
+
+    def node_ids(self) -> Iterable[int]:
+        return self._map.keys()

codeanalyzer/dataflow/syntactic.py

@@ -0,0 +1,26 @@
+################################################################################
+# Copyright IBM Corporation 2025
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#       http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+"""The L3 (syntactic) alias oracle: two access paths alias iff they are the
+identical path. Bypasses the type-based may-alias so def-use yields only
+name-equality (textual) edges — the alias-derived edges are the L4 delta."""
+
+from __future__ import annotations
+
+
+class SyntacticOracle:
+    def may_alias(self, path_a: str, path_b: str) -> bool:
+        return path_a == path_b