feat(dataflow): stage 5 — alias oracle wiring, Tarjan SCC, global qualification

Iterative Tarjan SCC condensation of the frozen call-graph oracle
(reverse topological schedule for bottom-up summaries); call mutations
become suffixed weak defs so caller-visible mutation is distinguishable
from local rebinding; global bases gain module::name qualification for
the interprocedural build. (#67)

Author: rahlk

codeanalyzer/dataflow/access_paths.py

@@ -372,16 +372,19 @@ def target_reads(self, target: ast.expr) -> Set[str]:
     # -- call mutation (documented over-approximation) -----------------------
 
     def mutation_defs(self, expr: ast.expr) -> Set[str]:
+        """Weak defs of the *contents* of receiver/argument objects (``xs.*``
+        — suffixed, so a call mutation is never confused with a local
+        rebinding, which is not caller-visible)."""
         defs: Set[str] = set()
         for call in _calls_in(expr):
             if isinstance(call.func, ast.Attribute):
                 receiver = self.path_of(call.func.value)
                 if receiver is not None:
-                    defs.add(receiver)
+                    defs.add(k_limit(receiver + ".*", self.k))
             for arg in list(call.args) + [kw.value for kw in call.keywords]:
                 p = self.path_of(arg)
                 if p is not None:
-                    defs.add(p)
+                    defs.add(k_limit(p + ".*", self.k))
         return defs
 
     def receiver_uses(self, expr: ast.expr) -> Set[str]:
@@ -419,12 +422,38 @@ def _calls_in(expr: ast.expr) -> List[ast.Call]:
     return calls
 
 
+def qualify_globals(paths: Set[str], scope: FunctionScope, qualifier: str) -> Set[str]:
+    """Rewrite global bases to their module-qualified form ``module::name``
+    (``::`` keeps the qualifier out of the field-path grammar). Builtins stay
+    bare — they carry no cross-module dataflow worth modeling."""
+    import builtins as _builtins
+
+    out: Set[str] = set()
+    for p in paths:
+        b = base_of(p)
+        if (
+            "::" not in b
+            and b != RETURN_PATH
+            and scope.kind_of(b) == "global"
+            and not hasattr(_builtins, b)
+        ):
+            out.add(f"{qualifier}::{b}" + p[len(b):])
+        else:
+            out.add(p)
+    return out
+
+
 def statement_facts(
-    cfg: ControlFlowGraph, func: ast.AST, scope: FunctionScope, k: int
+    cfg: ControlFlowGraph,
+    func: ast.AST,
+    scope: FunctionScope,
+    k: int,
+    global_qualifier: Optional[str] = None,
 ) -> Dict[int, StatementFacts]:
     """Defs/uses per CFG node id. Compound statements contribute only their
     header expressions; ENTRY defines every param/self/global/capture base
-    the function touches (the incoming state)."""
+    the function touches (the incoming state). With ``global_qualifier`` set
+    (the interprocedural build), global bases become ``module::name``."""
     ex = _PathExtractor(scope, k)
     facts: Dict[int, StatementFacts] = {}
 
@@ -526,6 +555,9 @@ def call_fx(expr: ast.expr) -> None:
 
         f.defs = {k_limit(p, k) for p in f.defs}
         f.uses = {k_limit(p, k) for p in f.uses}
+        if global_qualifier is not None:
+            f.defs = qualify_globals(f.defs, scope, global_qualifier)
+            f.uses = qualify_globals(f.uses, scope, global_qualifier)
         facts[node.id] = f
 
     return facts

codeanalyzer/dataflow/pdg.py

@@ -65,11 +65,12 @@ def build_pdg(
     enclosing_locals: Set[str],
     oracle: TypeBasedAliasOracle,
     k: int = 3,
+    global_qualifier: Optional[str] = None,
 ) -> 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)
+    facts = statement_facts(cfg, func, scope, k, global_qualifier)
 
     edges: List[PDGEdge] = [
         PDGEdge(source=a, target=b, type="CDG") for a, b in control_dependence(cfg)

codeanalyzer/dataflow/scc.py

@@ -0,0 +1,91 @@
+################################################################################
+# 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 5b of the level-3 dataflow ladder: SCC condensation of the call graph.
+
+The call graph is a frozen oracle (level-1 Jedi edges, provenance-merged with
+level-2 PyCG when enabled); Tarjan condenses it into strongly connected
+components, and the condensation DAG in reverse topological order is the
+bottom-up processing schedule for summary composition — callees before
+callers, one monotone fixpoint per SCC (mutual recursion).
+
+Iterative Tarjan (no recursion — real projects overflow Python's stack), with
+sorted tie-breaking so the schedule is deterministic.
+"""
+
+from __future__ import annotations
+
+from typing import Dict, List, Set, Tuple
+
+
+def strongly_connected_components(
+    nodes: List[str], edges: List[Tuple[str, str]]
+) -> List[List[str]]:
+    """Tarjan SCCs in reverse topological order (callees before callers).
+    Deterministic: nodes are visited in sorted order and members sorted."""
+    adj: Dict[str, List[str]] = {n: [] for n in nodes}
+    for s, t in sorted(set(edges)):
+        if s in adj and t in adj:
+            adj[s].append(t)
+
+    index_of: Dict[str, int] = {}
+    lowlink: Dict[str, int] = {}
+    on_stack: Set[str] = set()
+    stack: List[str] = []
+    sccs: List[List[str]] = []
+    counter = [0]
+
+    for root in sorted(adj):
+        if root in index_of:
+            continue
+        # Iterative DFS: (node, iterator position over successors).
+        work: List[Tuple[str, int]] = [(root, 0)]
+        while work:
+            node, i = work.pop()
+            if i == 0:
+                index_of[node] = lowlink[node] = counter[0]
+                counter[0] += 1
+                stack.append(node)
+                on_stack.add(node)
+            recurse = False
+            successors = adj[node]
+            while i < len(successors):
+                succ = successors[i]
+                i += 1
+                if succ not in index_of:
+                    work.append((node, i))
+                    work.append((succ, 0))
+                    recurse = True
+                    break
+                if succ in on_stack:
+                    lowlink[node] = min(lowlink[node], index_of[succ])
+            if recurse:
+                continue
+            if lowlink[node] == index_of[node]:
+                component: List[str] = []
+                while True:
+                    member = stack.pop()
+                    on_stack.discard(member)
+                    component.append(member)
+                    if member == node:
+                        break
+                sccs.append(sorted(component))
+            if work:
+                parent = work[-1][0]
+                lowlink[parent] = min(lowlink[parent], lowlink[node])
+
+    # Tarjan emits SCCs in reverse topological order already.
+    return sccs

test/test_dataflow_scc.py

@@ -0,0 +1,30 @@
+"""Stage-5 gate: SCC condensation of the call-graph oracle."""
+
+from codeanalyzer.dataflow.scc import strongly_connected_components
+
+
+def test_mutual_recursion_forms_one_scc():
+    nodes = ["main", "even", "odd", "leaf"]
+    edges = [("main", "even"), ("even", "odd"), ("odd", "even"), ("even", "leaf")]
+    sccs = strongly_connected_components(nodes, edges)
+    assert ["even", "odd"] in sccs
+    assert ["leaf"] in sccs and ["main"] in sccs
+
+
+def test_reverse_topological_order_callees_first():
+    nodes = ["a", "b", "c"]
+    edges = [("a", "b"), ("b", "c")]
+    sccs = strongly_connected_components(nodes, edges)
+    pos = {tuple(s): i for i, s in enumerate(sccs)}
+    assert pos[("c",)] < pos[("b",)] < pos[("a",)]
+
+
+def test_deterministic_across_runs():
+    nodes = ["m", "x", "y", "z"]
+    edges = [("m", "x"), ("x", "y"), ("y", "x"), ("y", "z"), ("z", "y")]
+    assert strongly_connected_components(nodes, edges) == strongly_connected_components(
+        nodes, edges
+    )
+    # x-y-z all collapse into one SCC (x↔y, y↔z), members sorted.
+    sccs = strongly_connected_components(nodes, edges)
+    assert ["x", "y", "z"] in sccs