feat(dataflow): stage 4 — PDG assembly and exact backward-slice gate

PDG = CDG ∪ DDG per callable over the same node ids; intraprocedural
backward slice as reverse reachability. Gate pins hand-computed exact
slices: the early-return arm is excluded from the other arm's slice,
loop slices close over the loop-carried dependency. (#67)

Author: rahlk

codeanalyzer/dataflow/pdg.py

@@ -0,0 +1,99 @@
+################################################################################
+# 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.
+################################################################################
+
+"""Stage 4 of the level-3 dataflow ladder: PDG assembly.
+
+Per callable, the PDG is the union of the stage-2 control-dependence edges
+(``CDG``) and the stage-3 def-use edges (``DDG``), over the same
+``(signature, node_id)`` nodes. Nothing new is computed here — this module is
+bookkeeping plus the intraprocedural backward slice that gates it: reverse
+reachability over CDG ∪ DDG from a criterion node, expected to match a
+hand-computed node set exactly on the fixture.
+"""
+
+from __future__ import annotations
+
+import ast
+from dataclasses import dataclass, field
+from typing import Dict, List, Optional, Set
+
+from codeanalyzer.dataflow.access_paths import (
+    FunctionScope,
+    StatementFacts,
+    build_scope,
+    statement_facts,
+)
+from codeanalyzer.dataflow.alias import TypeBasedAliasOracle
+from codeanalyzer.dataflow.cfg import ControlFlowGraph, build_cfg
+from codeanalyzer.dataflow.defuse import ddg_edges
+from codeanalyzer.dataflow.dominance import control_dependence
+
+
+@dataclass(frozen=True)
+class PDGEdge:
+    source: int
+    target: int
+    type: str  # "CDG" | "DDG"
+    var: Optional[str] = None  # access path on DDG edges
+
+
+@dataclass
+class FunctionPDG:
+    """One callable's intraprocedural graphs, keyed externally by signature."""
+
+    cfg: ControlFlowGraph
+    edges: List[PDGEdge]
+    scope: FunctionScope
+    facts: Dict[int, StatementFacts] = field(default_factory=dict)
+
+
+def build_pdg(
+    func: ast.AST,
+    enclosing_locals: Set[str],
+    oracle: TypeBasedAliasOracle,
+    k: int = 3,
+) -> FunctionPDG:
+    """CFG → dominance → def-use → PDG for one callable."""
+    cfg = build_cfg(func)
+    scope = build_scope(func, enclosing_locals)
+    facts = statement_facts(cfg, func, scope, k)
+
+    edges: List[PDGEdge] = [
+        PDGEdge(source=a, target=b, type="CDG") for a, b in control_dependence(cfg)
+    ]
+    edges.extend(
+        PDGEdge(source=e.source, target=e.target, type="DDG", var=e.var)
+        for e in ddg_edges(cfg, facts, oracle)
+    )
+    edges.sort(key=lambda e: (e.source, e.target, e.type, e.var or ""))
+    return FunctionPDG(cfg=cfg, edges=edges, scope=scope, facts=facts)
+
+
+def intraprocedural_backward_slice(pdg: FunctionPDG, criterion: int) -> Set[int]:
+    """Reverse reachability over CDG ∪ DDG from the criterion node (the
+    criterion itself is in the slice). The stage-4 gate."""
+    reverse: Dict[int, List[int]] = {}
+    for e in pdg.edges:
+        reverse.setdefault(e.target, []).append(e.source)
+    seen: Set[int] = set()
+    stack = [criterion]
+    while stack:
+        n = stack.pop()
+        if n in seen:
+            continue
+        seen.add(n)
+        stack.extend(reverse.get(n, []))
+    return seen

test/test_dataflow_pdg.py

@@ -0,0 +1,68 @@
+"""Stage-4 gate: PDG assembly and the exact intraprocedural backward slice.
+
+The highest-value test of the intraprocedural half: the backward slice of a
+named variable at a named line equals a hand-computed node set — exactly.
+It catches both missing control dependences and missing def-use edges.
+"""
+
+import ast
+from pathlib import Path
+
+from codeanalyzer.dataflow.alias import TypeBasedAliasOracle
+from codeanalyzer.dataflow.pdg import build_pdg, intraprocedural_backward_slice
+
+FIXTURE = Path(__file__).parent / "fixtures" / "single_functionalities" / "dataflow"
+
+
+def _pdg_of(file_name: str, func_name: str):
+    tree = ast.parse((FIXTURE / file_name).read_text())
+    for node in ast.walk(tree):
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == func_name:
+            return build_pdg(node, enclosing_locals=set(), oracle=TypeBasedAliasOracle())
+    raise AssertionError(f"{func_name} not found")
+
+
+def _id_at_line(pdg, line: int) -> int:
+    return next(n.id for n in pdg.cfg.nodes if n.start_line == line and n.kind != "entry")
+
+
+def _lines(pdg, ids) -> set:
+    return {
+        pdg.cfg.node_by_id(i).start_line
+        for i in ids
+        if pdg.cfg.node_by_id(i).kind not in ("entry", "exit")
+    }
+
+
+def test_pdg_edges_use_only_cdg_and_ddg_types():
+    for func in ("branchy", "looped", "early_exit", "handles"):
+        pdg = _pdg_of("main.py", func)
+        assert {e.type for e in pdg.edges} <= {"CDG", "DDG"}
+        for e in pdg.edges:
+            assert (e.var is not None) == (e.type == "DDG")
+
+
+def test_early_exit_slice_excludes_the_other_arm():
+    pdg = _pdg_of("main.py", "early_exit")
+    criterion = _id_at_line(pdg, 32)  # return y
+    slice_ids = intraprocedural_backward_slice(pdg, criterion)
+    # Hand-computed: ENTRY, the branch header (29), y = n * 2 (31), and the
+    # criterion itself. `return -1` (30) is control-dependent on the same
+    # branch but contributes nothing to y — it must NOT appear.
+    assert _lines(pdg, slice_ids) == {29, 31, 32}
+    assert pdg.cfg.entry_id in slice_ids
+    assert _id_at_line(pdg, 30) not in slice_ids
+
+
+def test_branchy_slice_includes_both_arms_and_the_branch():
+    pdg = _pdg_of("main.py", "branchy")
+    criterion = _id_at_line(pdg, 16)  # return x
+    slice_ids = intraprocedural_backward_slice(pdg, criterion)
+    assert _lines(pdg, slice_ids) == {12, 13, 15, 16}
+
+
+def test_looped_slice_of_return_total_is_the_whole_loop():
+    pdg = _pdg_of("main.py", "looped")
+    criterion = _id_at_line(pdg, 25)  # return total
+    slice_ids = intraprocedural_backward_slice(pdg, criterion)
+    assert _lines(pdg, slice_ids) == {20, 21, 22, 23, 24, 25}